systemprompt_loader/subprocess/
control.rs

1//! Async process control: liveness, ownership, port holders, termination.
2//!
3//! Everything a supervisor needs to decide whether a recorded pid is still
4//! one of ours and to stop it. Blocking work — identity reads that retry,
5//! `lsof`/`netstat`, every Windows primitive — runs under `spawn_blocking`;
6//! the POSIX signal and reap syscalls are non-blocking and run inline.
7//!
8//! A child this process spawned and that has since exited stays a zombie until
9//! it is reaped, and `kill(pid, 0)` reports a zombie as alive. Liveness here
10//! therefore reaps an exited child of ours first and then treats a zombie as
11//! exited, so a stop never waits out its grace period on a corpse.
12//!
13//! Copyright (c) systemprompt.io — Business Source License 1.1.
14//! See <https://systemprompt.io> for licensing details.
15
16use std::time::Duration;
17
18use systemprompt_identifiers::ServiceName;
19
20use super::{ChildKind, SupervisionError};
21
22#[cfg(unix)]
23use super::posix as platform;
24#[cfg(windows)]
25use super::winnt as platform;
26
27const EXIT_POLL: Duration = Duration::from_millis(25);
28const KILL_SETTLE: Duration = Duration::from_secs(5);
29
30/// How a termination request ended.
31#[derive(Debug, Clone, Copy, PartialEq, Eq)]
32pub enum Termination {
33    AlreadyExited,
34    Exited,
35    Killed,
36}
37
38/// How a stop of a recorded, marker-verified child ended.
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40pub enum StopOutcome {
41    NotRunning,
42    NotOurs,
43    Stopped(Termination),
44}
45
46pub async fn is_running(pid: u32) -> bool {
47    blocking_on_windows(move || !platform::has_exited(pid))
48        .await
49        .unwrap_or(false)
50}
51
52pub async fn owns(pid: u32, kind: ChildKind, service: &ServiceName) -> bool {
53    let service = service.clone();
54    let verdict = tokio::task::spawn_blocking(move || {
55        !platform::has_exited(pid)
56            && super::live_environ(pid).is_some_and(|environ| kind.identifies(&environ, &service))
57    })
58    .await;
59    match verdict {
60        Ok(owned) => owned,
61        Err(e) => {
62            tracing::warn!(pid, error = %e, "Child identity check did not complete; treating as not ours");
63            false
64        },
65    }
66}
67
68pub async fn pids_listening_on(port: u16) -> Result<Vec<u32>, SupervisionError> {
69    if port == 0 {
70        return Ok(Vec::new());
71    }
72    tokio::task::spawn_blocking(move || super::ports::listening_pids(port)).await?
73}
74
75pub async fn process_group(pid: u32) -> Option<u32> {
76    blocking_on_windows(move || platform::process_group(pid))
77        .await
78        .ok()
79        .flatten()
80}
81
82pub async fn terminate_gracefully(
83    pid: u32,
84    grace: Duration,
85) -> Result<Termination, SupervisionError> {
86    terminate(pid, grace, false).await
87}
88
89pub async fn terminate_group_gracefully(
90    pid: u32,
91    grace: Duration,
92) -> Result<Termination, SupervisionError> {
93    let leader = blocking_on_windows(move || platform::is_group_leader(pid)).await?;
94    terminate(pid, grace, leader).await
95}
96
97pub async fn stop_owned(
98    pid: u32,
99    kind: ChildKind,
100    service: &ServiceName,
101    grace: Duration,
102) -> Result<StopOutcome, SupervisionError> {
103    if !is_running(pid).await {
104        return Ok(StopOutcome::NotRunning);
105    }
106    if !owns(pid, kind, service).await {
107        tracing::warn!(
108            pid,
109            service = %service,
110            "Recorded pid is alive but carries no matching child marker; leaving it untouched"
111        );
112        return Ok(StopOutcome::NotOurs);
113    }
114    terminate_group_gracefully(pid, grace)
115        .await
116        .map(StopOutcome::Stopped)
117}
118
119async fn terminate(
120    pid: u32,
121    grace: Duration,
122    group: bool,
123) -> Result<Termination, SupervisionError> {
124    if !is_running(pid).await {
125        return Ok(Termination::AlreadyExited);
126    }
127    let delivery = blocking_on_windows(move || platform::send(pid, false, group)).await??;
128    if delivery == platform::Delivery::NoSuchProcess || wait_for_exit(pid, grace).await {
129        return Ok(Termination::Exited);
130    }
131    let delivery = blocking_on_windows(move || platform::send(pid, true, group)).await??;
132    if delivery == platform::Delivery::NoSuchProcess || wait_for_exit(pid, KILL_SETTLE).await {
133        return Ok(Termination::Killed);
134    }
135    Err(SupervisionError::Survived { pid })
136}
137
138async fn wait_for_exit(pid: u32, within: Duration) -> bool {
139    let deadline = tokio::time::Instant::now() + within;
140    loop {
141        if !is_running(pid).await {
142            return true;
143        }
144        if tokio::time::Instant::now() >= deadline {
145            return false;
146        }
147        tokio::time::sleep(EXIT_POLL).await;
148    }
149}
150
151#[cfg(unix)]
152fn blocking_on_windows<T, F>(work: F) -> impl Future<Output = Result<T, SupervisionError>>
153where
154    F: FnOnce() -> T + Send + 'static,
155    T: Send + 'static,
156{
157    std::future::ready(Ok(work()))
158}
159
160#[cfg(windows)]
161fn blocking_on_windows<T, F>(work: F) -> impl Future<Output = Result<T, SupervisionError>>
162where
163    F: FnOnce() -> T + Send + 'static,
164    T: Send + 'static,
165{
166    async move { Ok(tokio::task::spawn_blocking(work).await?) }
167}
systemprompt_loader/subprocess/control.rs

systemprompt_loader/subprocess/
control.rs
