rightkit_daemon/
incumbent.rs

1//! Replace a live but unresponsive incumbent host: ask it to stop, wait, then signal only a
2//! process whose identity matches the record (rightkit#1121).
3//!
4//! Takeover by rank ([`publish_takeover`](crate::publish_takeover)) is cooperative: the incumbent
5//! yields. A hung or old-version host never reads the request. [`stop_incumbent`] is the next
6//! step for a new host that is entitled to replace it: the caller's own `ask` (for a loopback HTTP
7//! daemon, `POST /internal/takeover`), a bounded wait, a graceful signal, a bounded wait and only
8//! then a forced stop of the tree. Every signal goes through
9//! [`AdoptedIdentity`](rightkit_child_process::AdoptedIdentity), which refuses a pid whose start
10//! time differs from the one recorded, so a foreign or recycled pid is never signalled. A record
11//! without a process start time ([`ServiceRecord::process_started_ms`]) cannot name one process:
12//! the incumbent is asked and waited for, and never signalled.
13
14use crate::record::ServiceRecord;
15use rightkit_child_process::AdoptedIdentity;
16use std::io;
17use std::num::NonZeroU32;
18use std::time::Duration;
19
20/// How far the recorded start time may differ from the live process's start time. The record
21/// stores milliseconds; the process table is exact.
22const START_TOLERANCE: Duration = Duration::from_secs(2);
23
24/// How the incumbent ended.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26#[non_exhaustive]
27pub enum IncumbentStop {
28    /// No process had the recorded pid already; nothing was asked or signalled.
29    AlreadyGone,
30    /// It stopped after being asked, within the grace.
31    Asked,
32    /// It stopped after one graceful signal.
33    Signalled,
34    /// It stopped after the forced stop of its tree.
35    Forced,
36}
37
38/// Why the incumbent was not replaced. Nothing was signalled for `NotIncumbent`, `NoIdentity`
39/// and `OwnProcess`.
40#[derive(Debug)]
41#[non_exhaustive]
42pub enum IncumbentError {
43    /// The pid is alive but is not the recorded process (a recycled pid): never signalled.
44    NotIncumbent { pid: u32 },
45    /// The record carries no process start time, so no signal may be sent; the incumbent was
46    /// asked and did not stop within the grace.
47    NoIdentity { pid: u32 },
48    /// The record names this process.
49    OwnProcess,
50    /// The incumbent is still running after the forced stop.
51    Survived { pid: u32 },
52    /// The OS refused a query or a signal.
53    Io(io::Error),
54}
55
56impl std::fmt::Display for IncumbentError {
57    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
58        match self {
59            Self::NotIncumbent { pid } => {
60                write!(
61                    f,
62                    "pid {pid} is not the recorded host (recycled); not signalled"
63                )
64            }
65            Self::NoIdentity { pid } => write!(
66                f,
67                "the record of pid {pid} has no start time, so it was asked but never signalled"
68            ),
69            Self::OwnProcess => write!(f, "the record names this process"),
70            Self::Survived { pid } => write!(f, "pid {pid} survived the forced stop"),
71            Self::Io(e) => write!(f, "{e}"),
72        }
73    }
74}
75
76impl std::error::Error for IncumbentError {}
77
78impl From<io::Error> for IncumbentError {
79    fn from(e: io::Error) -> Self {
80        Self::Io(e)
81    }
82}
83
84/// Stop the host `record` describes. `ask` is the app's polite request (its result is ignored: a
85/// hung host fails it, and the wait below decides), `grace` bounds each of the three waits
86/// (after the ask, after the graceful signal, after the forced stop).
87pub fn stop_incumbent(
88    record: &ServiceRecord,
89    grace: Duration,
90    ask: impl FnOnce() -> io::Result<()>,
91) -> Result<IncumbentStop, IncumbentError> {
92    let pid = record.pid;
93    if pid == std::process::id() {
94        return Err(IncumbentError::OwnProcess);
95    }
96    let Some(nz) = NonZeroU32::new(pid) else {
97        return Ok(IncumbentStop::AlreadyGone);
98    };
99    if !crate::transport::pid_alive(pid) {
100        return Ok(IncumbentStop::AlreadyGone);
101    }
102    if record.process_started_ms == 0 {
103        // No identity: ask and wait, never signal.
104        let _ = ask();
105        let deadline = std::time::Instant::now() + grace;
106        while std::time::Instant::now() < deadline {
107            if !crate::transport::pid_alive(pid) {
108                return Ok(IncumbentStop::Asked);
109            }
110            std::thread::sleep(Duration::from_millis(25));
111        }
112        return Err(IncumbentError::NoIdentity { pid });
113    }
114    let identity =
115        AdoptedIdentity::from_recorded_unix_ms(nz, record.process_started_ms, START_TOLERANCE)?;
116    let Some(identity) = identity else {
117        // Gone between the liveness probe and the identity read, or a recycled pid.
118        return if crate::transport::pid_alive(pid) {
119            Err(IncumbentError::NotIncumbent { pid })
120        } else {
121            Ok(IncumbentStop::AlreadyGone)
122        };
123    };
124    let _ = ask();
125    if identity.wait_confirmed_gone(grace)? {
126        return Ok(IncumbentStop::Asked);
127    }
128    if identity.quit_gracefully(grace)? {
129        return Ok(IncumbentStop::Signalled);
130    }
131    let report = identity.terminate_tree(grace)?;
132    if report.survivors.is_empty() && report.failed.is_empty() && identity.confirmed_gone()? {
133        Ok(IncumbentStop::Forced)
134    } else {
135        Err(IncumbentError::Survived { pid })
136    }
137}
rightkit_daemon/incumbent.rs

rightkit_daemon/
incumbent.rs
