// Copyright 2026 TiKV Project Authors. Licensed under Apache-2.0.

#[cfg(target_os = "linux")]
use std::os::unix::process::CommandExt;
use std::{
    fmt, io,
    os::unix::process::ExitStatusExt,
    path::{Path, PathBuf},
    process::{Child, Command, ExitStatus},
    sync::{
        Arc,
        atomic::{AtomicU32, Ordering},
        mpsc::{self, Receiver, RecvTimeoutError, SyncSender},
    },
    thread::{self, JoinHandle},
    time::{Duration, Instant},
};

#[cfg(target_os = "linux")]
use nix::libc;
use nix::{
    errno::Errno,
    libc::pid_t,
    sys::signal::{Signal, kill},
    unistd::Pid,
};

use crate::sys::thread::StdThreadBuildWrapper;

const POLL_INTERVAL: Duration = Duration::from_millis(100);
const STOP_POLL_INTERVAL: Duration = Duration::from_millis(50);
const STOP_TIMEOUT: Duration = Duration::from_secs(10);
const FORCE_KILL_REAP_TIMEOUT: Duration = Duration::from_secs(1);

/// Result type returned by supervised child-process operations.
pub type SupervisorResult<T> = std::result::Result<T, ProcessFailure>;

/// Timing configuration for the child watcher loop.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ChildSupervisorConfig {
    /// How long the watcher waits between liveness polls while the child is
    /// running.
    pub poll_interval: Duration,
    /// How long the watcher waits between stop-path polls after sending
    /// `SIGTERM`.
    pub stop_poll_interval: Duration,
    /// How long a graceful stop may take before the watcher escalates to
    /// `SIGKILL`.
    pub stop_timeout: Duration,
    /// How long the watcher waits for reaping after a forced `SIGKILL`.
    pub force_kill_reap_timeout: Duration,
}

impl Default for ChildSupervisorConfig {
    fn default() -> Self {
        Self {
            poll_interval: POLL_INTERVAL,
            stop_poll_interval: STOP_POLL_INTERVAL,
            stop_timeout: STOP_TIMEOUT,
            force_kill_reap_timeout: FORCE_KILL_REAP_TIMEOUT,
        }
    }
}

/// The caller-initiated action that caused a child stop.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum StopReason {
    Destroy,
    Restart,
}

impl fmt::Display for StopReason {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(match self {
            Self::Destroy => "destroy",
            Self::Restart => "restart",
        })
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ChildState {
    Running,
    Stopping { reason: StopReason },
}

/// A normalized view of an observed child exit status.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ObservedExit {
    Code(i32),
    Signal(i32),
    Unknown,
}

impl fmt::Display for ObservedExit {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Code(code) => write!(f, "code {code}"),
            Self::Signal(signal) => write!(f, "signal {signal}"),
            Self::Unknown => f.write_str("unknown status"),
        }
    }
}

/// Failure categories surfaced by the child supervisor.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ProcessFailureKind {
    UnexpectedExit {
        status: ObservedExit,
    },
    BadStopExit {
        reason: StopReason,
        status: ObservedExit,
    },
    StopTimeout {
        reason: StopReason,
        timeout: Duration,
    },
    SpawnFailed {
        error: String,
        command: String,
    },
    WatcherThreadSpawnFailed {
        error: String,
        thread_name: String,
    },
    SignalFailed {
        reason: StopReason,
        signal: &'static str,
        error: String,
    },
    WaitFailed {
        op: &'static str,
        error: String,
    },
    ChannelClosed {
        op: &'static str,
    },
    JoinFailed {
        op: &'static str,
    },
}

impl fmt::Display for ProcessFailureKind {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnexpectedExit { status } => write!(f, "unexpected exit with {status}"),
            Self::BadStopExit { reason, status } => {
                write!(f, "{reason} exited with {status}")
            }
            Self::StopTimeout { reason, timeout } => {
                write!(f, "{reason} timed out after {timeout:?}; forced SIGKILL")
            }
            Self::SpawnFailed { error, command } => {
                write!(f, "failed to spawn {command}: {error}")
            }
            Self::WatcherThreadSpawnFailed { error, thread_name } => {
                write!(f, "failed to spawn watcher thread {thread_name}: {error}")
            }
            Self::SignalFailed {
                reason,
                signal,
                error,
            } => write!(f, "failed to send {signal} during {reason}: {error}"),
            Self::WaitFailed { op, error } => write!(f, "{op} failed: {error}"),
            Self::ChannelClosed { op } => write!(f, "channel closed while {op}"),
            Self::JoinFailed { op } => write!(f, "watcher join failed while {op}"),
        }
    }
}

/// Rich error details for supervised child-process failures.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ProcessFailure {
    owner: &'static str,
    process_name: String,
    pid: Option<u32>,
    log_file: PathBuf,
    failure: ProcessFailureKind,
}

impl ProcessFailure {
    /// Returns the subsystem label attached to the supervised child.
    pub fn owner(&self) -> &'static str {
        self.owner
    }

    /// Returns the logical process name reported in failures.
    pub fn process_name(&self) -> &str {
        &self.process_name
    }

    /// Returns the last observed child pid, if one was recorded.
    pub fn pid(&self) -> Option<u32> {
        self.pid
    }

    /// Returns the child log file path associated with the failure.
    pub fn log_file(&self) -> &Path {
        &self.log_file
    }

    /// Returns the structured failure category.
    pub fn failure_kind(&self) -> &ProcessFailureKind {
        &self.failure
    }
}

impl fmt::Display for ProcessFailure {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        writeln!(f, "{} process failure:", self.owner)?;
        writeln!(f, "- process: {}", self.process_name)?;
        match self.pid {
            Some(pid) => writeln!(f, "- pid: {pid}")?,
            None => writeln!(f, "- pid: <unknown>")?,
        }
        writeln!(f, "- failure: {}", self.failure)?;
        write!(f, "- log_file: {}", self.log_file.display())
    }
}

/// Spawn parameters for a supervised helper process.
#[derive(Clone, Debug)]
pub struct ProcessSpawnSpec {
    owner: &'static str,
    process_name: String,
    thread_name: String,
    program: String,
    args: Vec<String>,
    log_file: PathBuf,
    config: ChildSupervisorConfig,
}

impl ProcessSpawnSpec {
    /// Builds a new supervised child-process specification.
    pub fn new(
        owner: &'static str,
        process_name: impl Into<String>,
        thread_name: impl Into<String>,
        program: String,
        args: Vec<String>,
        log_file: PathBuf,
    ) -> Self {
        Self {
            owner,
            process_name: process_name.into(),
            thread_name: thread_name.into(),
            program,
            args,
            log_file,
            config: ChildSupervisorConfig::default(),
        }
    }

    /// Returns the mutable supervisor timing configuration for this child.
    pub fn config_mut(&mut self) -> &mut ChildSupervisorConfig {
        &mut self.config
    }

    fn command_text(&self) -> String {
        let mut command = String::new();
        append_debug_arg(&mut command, &self.program);
        for arg in &self.args {
            append_debug_arg(&mut command, arg);
        }
        command
    }

    fn build_command(&self) -> Command {
        let mut cmd = Command::new(&self.program);
        cmd.args(&self.args);
        with_parent_death_signal(&mut cmd);
        cmd
    }
}

enum SupervisorCmd {
    Stop {
        reason: StopReason,
        reply: SyncSender<SupervisorResult<()>>,
    },
    Restart {
        reply: SyncSender<SupervisorResult<()>>,
    },
}

/// A synchronous control handle for a watcher thread that owns a child process.
///
/// After `spawn_supervisor()` succeeds, later supervisor failures are surfaced
/// by panicking in the watcher thread with a formatted [`ProcessFailure`].
/// Callers that need fail-fast behavior should install
/// `crate::set_panic_hook(false, ...)` before spawning the supervisor;
/// otherwise the original failure may later appear only as
/// [`ProcessFailureKind::ChannelClosed`] or [`ProcessFailureKind::JoinFailed`]
/// when the handle is used again.
pub struct ChildSupervisorHandle {
    owner: &'static str,
    process_name: String,
    tx: mpsc::Sender<SupervisorCmd>,
    join: JoinHandle<()>,
    log_file: PathBuf,
    pid: Arc<AtomicU32>,
}

impl ChildSupervisorHandle {
    /// Requests a graceful stop and waits for the watcher to acknowledge it.
    pub fn request_stop(&self, reason: StopReason) -> SupervisorResult<()> {
        let (reply_tx, reply_rx) = mpsc::sync_channel(0);
        self.tx
            .send(SupervisorCmd::Stop {
                reason,
                reply: reply_tx,
            })
            .map_err(|_| self.channel_closed("sending stop command"))?;
        reply_rx
            .recv()
            .map_err(|_| self.channel_closed("waiting for stop reply"))?
    }

    /// Requests an in-place restart of the supervised child.
    pub fn request_restart(&self) -> SupervisorResult<()> {
        let (reply_tx, reply_rx) = mpsc::sync_channel(0);
        self.tx
            .send(SupervisorCmd::Restart { reply: reply_tx })
            .map_err(|_| self.channel_closed("sending restart command"))?;
        reply_rx
            .recv()
            .map_err(|_| self.channel_closed("waiting for restart reply"))?
    }

    /// Requests a stop and then joins the watcher thread.
    pub fn stop_and_join(self, reason: StopReason) -> SupervisorResult<()> {
        let stop_res = self.request_stop(reason);
        let join_res = self.join();
        stop_res.and(join_res)
    }

    /// Joins the watcher thread after it has already been stopped.
    pub fn join(self) -> SupervisorResult<()> {
        let ChildSupervisorHandle {
            owner,
            process_name,
            join,
            log_file,
            pid,
            ..
        } = self;
        join.join().map_err(|_| ProcessFailure {
            owner,
            process_name,
            pid: pid_value(&pid),
            log_file,
            failure: ProcessFailureKind::JoinFailed {
                op: "joining watcher thread",
            },
        })
    }

    /// Returns the logical process name reported by this handle.
    pub fn process_name(&self) -> &str {
        &self.process_name
    }

    /// Returns the child log file path associated with this handle.
    pub fn log_file(&self) -> &Path {
        &self.log_file
    }

    /// Returns the last observed child pid, if one has been recorded.
    pub fn pid(&self) -> Option<u32> {
        pid_value(&self.pid)
    }

    fn channel_closed(&self, op: &'static str) -> ProcessFailure {
        ProcessFailure {
            owner: self.owner,
            process_name: self.process_name.clone(),
            pid: pid_value(&self.pid),
            log_file: self.log_file.clone(),
            failure: ProcessFailureKind::ChannelClosed { op },
        }
    }
}

struct ChildSupervisor {
    child: Child,
    state: ChildState,
    spawn_spec: ProcessSpawnSpec,
    pid: Arc<AtomicU32>,
}

impl ChildSupervisor {
    fn new(spawn_spec: ProcessSpawnSpec, child: Child, pid: Arc<AtomicU32>) -> Self {
        pid.store(child.id(), Ordering::Relaxed);
        Self {
            child,
            state: ChildState::Running,
            spawn_spec,
            pid,
        }
    }

    fn spawn(
        spawn_spec: ProcessSpawnSpec,
        pid: Arc<AtomicU32>,
    ) -> SupervisorResult<ChildSupervisor> {
        let child = spawn_child(&spawn_spec)?;
        Ok(Self::new(spawn_spec, child, pid))
    }

    fn run_loop(&mut self, rx: Receiver<SupervisorCmd>) -> SupervisorResult<()> {
        loop {
            match rx.recv_timeout(self.spawn_spec.config.poll_interval) {
                Ok(SupervisorCmd::Stop { reason, reply }) => {
                    let res = self.stop_child(reason);
                    let send_res = self.send_reply(reply, res.clone(), "sending stop reply");
                    match (res, send_res) {
                        (Ok(()), Ok(())) => return Ok(()),
                        (Err(err), _) => return Err(err),
                        (Ok(()), Err(err)) => return Err(err),
                    }
                }
                Ok(SupervisorCmd::Restart { reply }) => {
                    let res = self.restart_child();
                    let send_res = self.send_reply(reply, res.clone(), "sending restart reply");
                    match (res, send_res) {
                        (Ok(()), Ok(())) => {}
                        (Err(err), _) => return Err(err),
                        (Ok(()), Err(err)) => return Err(err),
                    }
                }
                Err(RecvTimeoutError::Timeout) => self.poll_child()?,
                Err(RecvTimeoutError::Disconnected) => {
                    return self.stop_after_command_disconnect();
                }
            }
        }
    }

    fn poll_child(&mut self) -> SupervisorResult<()> {
        let Some(status) = self
            .child
            .try_wait()
            .map_err(|err| self.wait_failed("try_wait", err))?
        else {
            return Ok(());
        };

        match self.state {
            ChildState::Running => Err(self.failure(ProcessFailureKind::UnexpectedExit {
                status: observe_exit(status),
            })),
            ChildState::Stopping { reason } => self.validate_stop_exit(reason, status),
        }
    }

    fn stop_child(&mut self, reason: StopReason) -> SupervisorResult<()> {
        // Snapshot liveness before SIGTERM so an already-dead child is still
        // reported as an unexpected exit. A narrow kernel race remains between
        // this check and kill(2).
        if let Some(status) = self
            .child
            .try_wait()
            .map_err(|err| self.wait_failed("try_wait before stop", err))?
        {
            return Err(self.failure(ProcessFailureKind::UnexpectedExit {
                status: observe_exit(status),
            }));
        }

        if let Err(err) = send_signal_to_child(&self.child, Signal::SIGTERM) {
            if err == Errno::ESRCH {
                let status = self
                    .child
                    .wait()
                    .map_err(|wait_err| self.wait_failed("wait after ESRCH", wait_err))?;
                return Err(self.failure(ProcessFailureKind::UnexpectedExit {
                    status: observe_exit(status),
                }));
            }
            return Err(self.failure(ProcessFailureKind::SignalFailed {
                reason,
                signal: "SIGTERM",
                error: err.to_string(),
            }));
        }

        self.state = ChildState::Stopping { reason };
        let deadline = Instant::now() + self.spawn_spec.config.stop_timeout;
        loop {
            match self
                .child
                .try_wait()
                .map_err(|err| self.wait_failed("try_wait during stop", err))?
            {
                Some(status) => return self.validate_stop_exit(reason, status),
                None if Instant::now() >= deadline => {
                    self.force_kill_and_reap_best_effort();
                    return Err(self.failure(ProcessFailureKind::StopTimeout {
                        reason,
                        timeout: self.spawn_spec.config.stop_timeout,
                    }));
                }
                None => thread::sleep(self.spawn_spec.config.stop_poll_interval),
            }
        }
    }

    fn restart_child(&mut self) -> SupervisorResult<()> {
        self.stop_child(StopReason::Restart)?;
        let child = spawn_child(&self.spawn_spec)?;
        self.pid.store(child.id(), Ordering::Relaxed);
        self.child = child;
        self.state = ChildState::Running;
        Ok(())
    }

    fn validate_stop_exit(&self, reason: StopReason, status: ExitStatus) -> SupervisorResult<()> {
        if status.success() {
            Ok(())
        } else {
            Err(self.failure(ProcessFailureKind::BadStopExit {
                reason,
                status: observe_exit(status),
            }))
        }
    }

    fn stop_after_command_disconnect(&mut self) -> SupervisorResult<()> {
        if let Err(err) = self.stop_child(StopReason::Destroy) {
            self.force_kill_and_reap_best_effort();
            return Err(err);
        }
        Ok(())
    }

    fn force_kill_and_reap_best_effort(&mut self) {
        let _ = send_signal_to_child(&self.child, Signal::SIGKILL);

        // Keep the timeout path bounded as well. A later parent exit will still
        // let init reap the child if the kernel never reports it.
        let deadline = Instant::now() + self.spawn_spec.config.force_kill_reap_timeout;
        while Instant::now() < deadline {
            match self.child.try_wait() {
                Ok(Some(_)) | Err(_) => break,
                Ok(None) => thread::sleep(self.spawn_spec.config.stop_poll_interval),
            }
        }
    }

    fn failure(&self, failure: ProcessFailureKind) -> ProcessFailure {
        ProcessFailure {
            owner: self.spawn_spec.owner,
            process_name: self.spawn_spec.process_name.clone(),
            pid: Some(self.child.id()),
            log_file: self.spawn_spec.log_file.clone(),
            failure,
        }
    }

    fn wait_failed(&self, op: &'static str, err: io::Error) -> ProcessFailure {
        self.failure(ProcessFailureKind::WaitFailed {
            op,
            error: err.to_string(),
        })
    }

    fn send_reply(
        &self,
        reply: SyncSender<SupervisorResult<()>>,
        res: SupervisorResult<()>,
        op: &'static str,
    ) -> SupervisorResult<()> {
        reply
            .send(res)
            .map_err(|_| self.failure(ProcessFailureKind::ChannelClosed { op }))
    }
}

/// Starts a child process and a watcher thread that owns it.
///
/// Any unexpected child exit is reported as a supervisor failure.
///
/// After startup, supervisor failures panic in the watcher thread instead of
/// being returned through [`ChildSupervisorHandle`]. Callers that need those
/// failures to terminate the process immediately should install
/// `crate::set_panic_hook(false, ...)` before spawning the supervisor.
///
/// On Linux, the child is configured with `PR_SET_PDEATHSIG=SIGKILL` so it
/// does not outlive a fail-fast parent exit. Other Unix targets skip this
/// best-effort parent-death hook because `prctl` is unavailable there.
pub fn spawn_supervisor(spawn_spec: ProcessSpawnSpec) -> SupervisorResult<ChildSupervisorHandle> {
    let owner = spawn_spec.owner;
    let process_name = spawn_spec.process_name.clone();
    let log_file = spawn_spec.log_file.clone();
    let pid = Arc::new(AtomicU32::new(0));
    let pid_for_thread = pid.clone();
    let (tx, rx) = mpsc::channel();
    let (ready_tx, ready_rx) = mpsc::sync_channel(0);
    let thread_name = spawn_spec.thread_name.clone();

    let join = thread::Builder::new()
        .name(thread_name.clone())
        .spawn_wrapper(move || {
            let mut supervisor = match ChildSupervisor::spawn(spawn_spec, pid_for_thread) {
                Ok(supervisor) => supervisor,
                Err(failure) => {
                    let _ = ready_tx.send(Err(failure));
                    return;
                }
            };
            if let Err(failure) = supervisor.poll_child() {
                let _ = ready_tx.send(Err(failure));
                return;
            }
            if ready_tx.send(Ok(())).is_err() {
                return;
            }
            if let Err(failure) = supervisor.run_loop(rx) {
                panic!("{}", failure);
            }
        })
        .map_err(|err| ProcessFailure {
            owner,
            process_name: process_name.clone(),
            pid: None,
            log_file: log_file.clone(),
            failure: ProcessFailureKind::WatcherThreadSpawnFailed {
                error: err.to_string(),
                thread_name,
            },
        })?;

    ready_rx.recv().map_err(|_| ProcessFailure {
        owner,
        process_name: process_name.clone(),
        pid: pid_value(&pid),
        log_file: log_file.clone(),
        failure: ProcessFailureKind::ChannelClosed {
            op: "waiting for watcher startup",
        },
    })??;

    Ok(ChildSupervisorHandle {
        owner,
        process_name,
        tx,
        join,
        log_file,
        pid,
    })
}

fn observe_exit(status: ExitStatus) -> ObservedExit {
    if let Some(code) = status.code() {
        ObservedExit::Code(code)
    } else if let Some(signal) = status.signal() {
        ObservedExit::Signal(signal)
    } else {
        ObservedExit::Unknown
    }
}

fn append_debug_arg(command: &mut String, arg: &str) {
    if !command.is_empty() {
        command.push(' ');
    }
    command.push_str(&format!("{arg:?}"));
}

fn pid_value(pid: &Arc<AtomicU32>) -> Option<u32> {
    let pid = pid.load(Ordering::Relaxed);
    (pid != 0).then_some(pid)
}

#[cfg(target_os = "linux")]
fn with_parent_death_signal(cmd: &mut Command) {
    let parent_pid = unsafe { libc::getpid() };
    unsafe {
        cmd.pre_exec(move || {
            if libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGKILL) != 0 {
                return Err(io::Error::last_os_error());
            }
            if libc::getppid() != parent_pid {
                return Err(io::Error::from_raw_os_error(libc::ESRCH));
            }
            Ok(())
        });
    }
}

#[cfg(not(target_os = "linux"))]
fn with_parent_death_signal(_cmd: &mut Command) {}

fn spawn_child(spawn_spec: &ProcessSpawnSpec) -> SupervisorResult<Child> {
    let cmd_text = spawn_spec.command_text();
    let mut cmd = spawn_spec.build_command();
    info!(
        "start supervised child";
        "owner" => spawn_spec.owner,
        "process" => &spawn_spec.process_name,
        "cmd" => &cmd_text
    );
    cmd.spawn().map_err(|err| ProcessFailure {
        owner: spawn_spec.owner,
        process_name: spawn_spec.process_name.clone(),
        pid: None,
        log_file: spawn_spec.log_file.clone(),
        failure: ProcessFailureKind::SpawnFailed {
            error: err.to_string(),
            command: cmd_text,
        },
    })
}

fn send_signal_to_child(child: &Child, signal: Signal) -> nix::Result<()> {
    let pid = Pid::from_raw(child.id() as pid_t);
    kill(pid, signal)
}

#[cfg(test)]
mod tests {
    use std::{env, process::Command, sync::atomic::AtomicU32};

    use tempfile::TempDir;

    use super::*;

    const PYTHON: &str = "python3";
    const FAILFAST_ENV: &str = "TIKV_UTIL_CHILD_SUPERVISOR_FAILFAST_SUBPROCESS";

    fn test_spawn_spec(process_name: &str, script: &str, dir: &TempDir) -> ProcessSpawnSpec {
        let mut spec = ProcessSpawnSpec::new(
            "local provider",
            process_name.to_string(),
            format!("test-{process_name}"),
            PYTHON.to_string(),
            vec!["-c".to_string(), script.to_string()],
            dir.path().join(format!("{process_name}.log")),
        );
        *spec.config_mut() = ChildSupervisorConfig {
            poll_interval: Duration::from_millis(10),
            stop_poll_interval: Duration::from_millis(10),
            stop_timeout: Duration::from_secs(1),
            force_kill_reap_timeout: Duration::from_secs(1),
        };
        spec
    }

    fn test_supervisor<F>(process_name: &str, build_script: F) -> (TempDir, ChildSupervisor)
    where
        F: FnOnce(&std::path::Path) -> String,
    {
        let dir = tempfile::Builder::new()
            .prefix("rep-local-provider-")
            .tempdir()
            .unwrap();
        let ready_file = dir.path().join(format!("{process_name}.ready"));
        let script = build_script(&ready_file);
        let spec = test_spawn_spec(process_name, &script, &dir);
        let child = spawn_child(&spec).unwrap();
        let pid = Arc::new(AtomicU32::new(child.id()));
        let supervisor = ChildSupervisor::new(spec, child, pid);
        wait_for_path(&ready_file, Duration::from_secs(1));
        (dir, supervisor)
    }

    fn wait_for_exit(supervisor: &mut ChildSupervisor, timeout: Duration) -> Option<ExitStatus> {
        let deadline = Instant::now() + timeout;
        while Instant::now() < deadline {
            match supervisor.child.try_wait().unwrap() {
                Some(status) => return Some(status),
                None => thread::sleep(Duration::from_millis(10)),
            }
        }
        None
    }

    fn graceful_term_script(exit_code: i32, ready_file: &std::path::Path) -> String {
        format!(
            concat!(
                "import signal, sys, time\n",
                "from pathlib import Path\n",
                "signal.signal(signal.SIGTERM, lambda *_: sys.exit({exit_code}))\n",
                "Path({ready_file:?}).write_text('ready')\n",
                "while True:\n",
                "    time.sleep(1)\n",
            ),
            exit_code = exit_code,
            ready_file = ready_file,
        )
    }

    fn exit_script(exit_code: i32, ready_file: &std::path::Path) -> String {
        format!(
            concat!(
                "from pathlib import Path\n",
                "import sys\n",
                "Path({ready_file:?}).write_text('ready')\n",
                "sys.exit({exit_code})\n",
            ),
            exit_code = exit_code,
            ready_file = ready_file,
        )
    }

    fn ignore_term_script(ready_file: &std::path::Path) -> String {
        format!(
            concat!(
                "import signal, time\n",
                "from pathlib import Path\n",
                "signal.signal(signal.SIGTERM, signal.SIG_IGN)\n",
                "Path({ready_file:?}).write_text('ready')\n",
                "while True:\n",
                "    time.sleep(1)\n",
            ),
            ready_file = ready_file,
        )
    }

    fn wait_for_path(path: &std::path::Path, timeout: Duration) {
        let deadline = Instant::now() + timeout;
        while Instant::now() < deadline {
            if path.exists() {
                return;
            }
            thread::sleep(Duration::from_millis(10));
        }
        panic!("timed out waiting for {}", path.display());
    }

    #[test]
    fn test_poll_child_reports_unexpected_exit_code_zero() {
        let (_dir, mut supervisor) = test_supervisor("rep-pd", |ready| exit_script(0, ready));
        wait_for_exit(&mut supervisor, Duration::from_secs(1)).unwrap();
        let err = supervisor.poll_child().unwrap_err();
        assert_eq!(
            err.failure,
            ProcessFailureKind::UnexpectedExit {
                status: ObservedExit::Code(0),
            }
        );
    }

    #[test]
    fn test_poll_child_reports_unexpected_exit_code_one() {
        let (_dir, mut supervisor) = test_supervisor("rep-cdc", |ready| exit_script(1, ready));
        wait_for_exit(&mut supervisor, Duration::from_secs(1)).unwrap();
        let err = supervisor.poll_child().unwrap_err();
        assert_eq!(
            err.failure,
            ProcessFailureKind::UnexpectedExit {
                status: ObservedExit::Code(1),
            }
        );
    }

    #[test]
    fn test_stop_child_accepts_exit_zero() {
        let (_dir, mut supervisor) =
            test_supervisor("rep-pd", |ready| graceful_term_script(0, ready));
        supervisor.stop_child(StopReason::Destroy).unwrap();
        assert!(wait_for_exit(&mut supervisor, Duration::from_secs(1)).is_some());
    }

    #[test]
    fn test_stop_child_rejects_exit_one() {
        for process_name in ["rep-cdc", "rep-pd", "rep-tidb"] {
            let (_dir, mut supervisor) =
                test_supervisor(process_name, |ready| graceful_term_script(1, ready));
            let err = supervisor.stop_child(StopReason::Destroy).unwrap_err();
            assert_eq!(
                err.failure,
                ProcessFailureKind::BadStopExit {
                    reason: StopReason::Destroy,
                    status: ObservedExit::Code(1),
                }
            );
        }
    }

    #[test]
    fn test_stop_child_reports_already_exited_process_as_unexpected() {
        let (_dir, mut supervisor) = test_supervisor("rep-tidb", |ready| exit_script(0, ready));
        wait_for_exit(&mut supervisor, Duration::from_secs(1)).unwrap();
        let err = supervisor.stop_child(StopReason::Destroy).unwrap_err();
        assert_eq!(
            err.failure,
            ProcessFailureKind::UnexpectedExit {
                status: ObservedExit::Code(0),
            }
        );
    }

    #[test]
    fn test_stop_child_times_out_and_forces_sigkill() {
        let (_dir, mut supervisor) = test_supervisor("rep-cdc", ignore_term_script);
        supervisor.spawn_spec.config.stop_timeout = Duration::from_millis(100);
        supervisor.spawn_spec.config.force_kill_reap_timeout = Duration::from_secs(1);
        let err = supervisor.stop_child(StopReason::Destroy).unwrap_err();
        assert_eq!(
            err.failure,
            ProcessFailureKind::StopTimeout {
                reason: StopReason::Destroy,
                timeout: Duration::from_millis(100),
            }
        );
        assert!(wait_for_exit(&mut supervisor, Duration::from_secs(1)).is_some());
    }

    #[test]
    fn test_restart_child_restarts_process_without_false_stop_error() {
        let (_dir, mut supervisor) =
            test_supervisor("rep-pd", |ready| graceful_term_script(0, ready));
        supervisor.restart_child().unwrap();
        assert_eq!(supervisor.state, ChildState::Running);
        assert!(supervisor.child.try_wait().unwrap().is_none());
        thread::sleep(Duration::from_millis(50));
        supervisor.stop_child(StopReason::Destroy).unwrap();
    }

    #[test]
    fn test_restart_child_reports_spawn_failure() {
        let (_dir, mut supervisor) =
            test_supervisor("rep-pd", |ready| graceful_term_script(0, ready));
        supervisor.spawn_spec.program = "/path/that/does/not/exist".to_string();
        let err = supervisor.restart_child().unwrap_err();
        match err.failure {
            ProcessFailureKind::SpawnFailed { .. } => {}
            other => panic!("unexpected failure: {:?}", other),
        }
    }

    #[test]
    fn test_request_stop_reports_channel_disconnect() {
        let (tx, reply_rx) = mpsc::channel::<SupervisorCmd>();
        drop(reply_rx);
        let handle = ChildSupervisorHandle {
            owner: "local provider",
            process_name: "rep-pd".to_string(),
            tx,
            join: thread::spawn(|| {}),
            log_file: PathBuf::from("/tmp/rep-pd.log"),
            pid: Arc::new(AtomicU32::new(42)),
        };
        let err = handle.request_stop(StopReason::Destroy).unwrap_err();
        assert_eq!(
            err.failure,
            ProcessFailureKind::ChannelClosed {
                op: "sending stop command",
            }
        );
        handle.join().unwrap();
    }

    #[test]
    fn test_run_loop_stops_child_after_command_disconnect() {
        let (_dir, mut supervisor) =
            test_supervisor("rep-pd", |ready| graceful_term_script(0, ready));
        let (tx, rx) = mpsc::channel::<SupervisorCmd>();
        drop(tx);

        supervisor.run_loop(rx).unwrap();
        assert!(wait_for_exit(&mut supervisor, Duration::from_secs(1)).is_some());
    }

    #[test]
    fn test_stop_and_join_preserves_stop_error_after_joining() {
        let (tx, rx) = mpsc::channel::<SupervisorCmd>();
        let joined = Arc::new(AtomicU32::new(0));
        let joined_for_thread = joined.clone();
        let join = thread::spawn(move || {
            match rx.recv().unwrap() {
                SupervisorCmd::Stop { reason, reply } => {
                    assert_eq!(reason, StopReason::Destroy);
                    reply
                        .send(Err(ProcessFailure {
                            owner: "local provider",
                            process_name: "rep-pd".to_string(),
                            pid: Some(42),
                            log_file: PathBuf::from("/tmp/rep-pd.log"),
                            failure: ProcessFailureKind::BadStopExit {
                                reason: StopReason::Destroy,
                                status: ObservedExit::Code(1),
                            },
                        }))
                        .unwrap();
                }
                SupervisorCmd::Restart { .. } => panic!("unexpected restart command"),
            }
            thread::sleep(Duration::from_millis(100));
            joined_for_thread.store(1, Ordering::Relaxed);
        });
        let handle = ChildSupervisorHandle {
            owner: "local provider",
            process_name: "rep-pd".to_string(),
            tx,
            join,
            log_file: PathBuf::from("/tmp/rep-pd.log"),
            pid: Arc::new(AtomicU32::new(42)),
        };

        let err = handle.stop_and_join(StopReason::Destroy).unwrap_err();
        assert_eq!(
            err.failure,
            ProcessFailureKind::BadStopExit {
                reason: StopReason::Destroy,
                status: ObservedExit::Code(1),
            }
        );
        assert_eq!(joined.load(Ordering::Relaxed), 1);
    }

    #[test]
    fn test_join_reports_watcher_panic() {
        let (tx, _rx) = mpsc::channel::<SupervisorCmd>();
        let handle = ChildSupervisorHandle {
            owner: "local provider",
            process_name: "rep-pd".to_string(),
            tx,
            join: thread::spawn(|| panic!("boom")),
            log_file: PathBuf::from("/tmp/rep-pd.log"),
            pid: Arc::new(AtomicU32::new(42)),
        };
        let err = handle.join().unwrap_err();
        assert_eq!(
            err.failure,
            ProcessFailureKind::JoinFailed {
                op: "joining watcher thread",
            }
        );
    }

    #[test]
    fn test_supervisor_fail_fast_subprocess() {
        if env::var_os(FAILFAST_ENV).is_some() {
            run_failfast_subprocess();
            unreachable!("panic hook should terminate the subprocess");
        }

        let output = Command::new(env::current_exe().unwrap())
            .arg("--nocapture")
            .arg("test_supervisor_fail_fast_subprocess")
            .env(FAILFAST_ENV, "1")
            .env("RUST_TEST_THREADS", "1")
            .output()
            .unwrap();
        assert!(!output.status.success());

        let mut combined = String::from_utf8_lossy(&output.stdout).into_owned();
        combined.push_str(&String::from_utf8_lossy(&output.stderr));
        assert!(combined.contains("local provider process failure"));
        assert!(combined.contains("process: rep-cdc"));
        assert!(combined.contains("unexpected exit"));
    }

    fn run_failfast_subprocess() {
        let dir = tempfile::Builder::new()
            .prefix("rep-local-provider-subprocess-")
            .tempdir()
            .unwrap();
        let drainer = crate::logger::text_format(crate::logger::term_writer(), true);
        let _ = crate::logger::init_log(
            drainer,
            crate::logger::get_level_by_string("debug").unwrap(),
            false,
            true,
            vec![],
            0,
        );
        crate::set_panic_hook(false, dir.path().to_str().unwrap());

        let ready_file = dir.path().join("rep-cdc.ready");
        let spec = test_spawn_spec("rep-cdc", &graceful_term_script(0, &ready_file), &dir);
        let handle = spawn_supervisor(spec).unwrap();
        wait_for_path(&ready_file, Duration::from_secs(1));
        let pid = handle.pid().unwrap();
        send_signal_to_child_pid(pid, Signal::SIGTERM).unwrap();

        thread::sleep(Duration::from_secs(10));
        panic!("watcher did not fail fast");
    }

    fn send_signal_to_child_pid(pid: u32, signal: Signal) -> nix::Result<()> {
        kill(Pid::from_raw(pid as pid_t), signal)
    }
}