wsb-daemon 0.248.6265

wsb's daemon: the workbench, the GitHub mirror's schedule and peer continuity, for every workspace on the machine
//! The children an operation is running, and stopping them.
//!
//! A run spawns dozens of processes and the page used to show them as a wall
//! of identical chips reading `bash` — the first word of every argv, which is
//! the one word they all share. What a person needs from a running child is
//! which one it is, how long it has been going, and a way to stop it.
//!
//! A pid arriving from a browser is not trusted. It is checked against this
//! operation's own log: a child it never started, or one it already reported
//! finished, is refused. That is what keeps "kill" scoped to the run being
//! looked at rather than to anything on the machine.

use axum::{
    extract::{Path as AxumPath, State},
    http::StatusCode,
    Json,
};
use wsb::output::Record;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use uuid::Uuid;

use crate::web::state::WebState;

/// One child still running.
#[derive(Debug, Clone, Serialize)]
pub struct RunningChild {
    /// The id the records use, which is unique within this operation.
    pub id: u64,
    pub pid: u32,
    pub argv: Vec<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cwd: Option<String>,
    /// When it was announced, RFC 3339, so a page can count up from it.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub started_at: Option<String>,
}

/// Every child this operation started and has not reported finished.
///
/// Read from the log rather than kept as separate state: the log is what the
/// page is showing, so an answer derived from anything else could disagree
/// with what the reader can see.
async fn running_children(state: &WebState, id: Uuid) -> Vec<RunningChild> {
    let Some((lines, _, _)) = state.ops.output_snapshot(id).await else {
        return Vec::new();
    };
    let mut running: BTreeMap<u64, RunningChild> = BTreeMap::new();
    for line in &lines {
        match &line.record {
            Record::SubprocessStarted {
                id: child,
                argv,
                cwd,
                pid,
            } => {
                running.insert(
                    *child,
                    RunningChild {
                        id: *child,
                        pid: *pid,
                        argv: argv.clone(),
                        cwd: cwd.clone(),
                        started_at: line.ts.clone(),
                    },
                );
            }
            Record::SubprocessFinished { id: child, .. } => {
                running.remove(child);
            }
            _ => {}
        }
    }
    running.into_values().collect()
}

pub async fn list(
    State(state): State<WebState>,
    AxumPath(id): AxumPath<Uuid>,
) -> Json<Vec<RunningChild>> {
    Json(running_children(&state, id).await)
}

#[derive(Debug, Deserialize)]
pub struct KillRequest {
    /// Which children to stop. Empty with `all` set means every one of them.
    #[serde(default)]
    pub pids: Vec<u32>,
    #[serde(default)]
    pub all: bool,
    /// `TERM` by default; `KILL` for one that will not go.
    #[serde(default)]
    pub signal: Option<String>,
}

#[derive(Debug, Serialize)]
pub struct KillOutcome {
    /// Processes that are no longer running.
    pub signalled: Vec<u32>,
    /// Pids that were asked for and refused, with why.
    pub refused: Vec<Refusal>,
}

#[derive(Debug, Serialize)]
pub struct Refusal {
    pub pid: u32,
    pub reason: String,
}

pub async fn kill(
    State(state): State<WebState>,
    AxumPath(id): AxumPath<Uuid>,
    Json(request): Json<KillRequest>,
) -> Result<Json<KillOutcome>, (StatusCode, Json<serde_json::Value>)> {
    let children = running_children(&state, id).await;
    let known: Vec<u32> = children.iter().map(|child| child.pid).collect();

    let asked: Vec<u32> = if request.all {
        known.clone()
    } else {
        request.pids.clone()
    };

    // A signal name is not passed through to a shell, but it does reach a
    // command line, and anything outside this pair is a caller asking for
    // something this endpoint does not offer.
    let signal = match request.signal.as_deref() {
        None | Some("TERM") => "TERM",
        Some("KILL") => "KILL",
        Some(other) => {
            return Err((
                StatusCode::BAD_REQUEST,
                Json(serde_json::json!({
                    "error": format!("signal must be TERM or KILL, not {other}"),
                })),
            ))
        }
    };

    let ours = wsb::ops::process::lineage();
    let mut signalled = Vec::new();
    let mut refused = Vec::new();
    for pid in asked {
        if !known.contains(&pid) {
            refused.push(Refusal {
                pid,
                reason: "not a child this operation is running".to_string(),
            });
            continue;
        }
        // Belt and braces. A pid this daemon started cannot be the daemon, but
        // pids are reused, and signalling our own lineage would take down the
        // workbench along with the child.
        if ours.contains(&pid) {
            refused.push(Refusal {
                pid,
                reason: "this is the daemon's own process".to_string(),
            });
            continue;
        }
        signalled.push(pid);
    }

    let mut stopped = Vec::new();
    if !signalled.is_empty() {
        let pids = signalled.clone();
        let named = signal.to_string();
        let results = tokio::task::spawn_blocking(move || {
            let sent = wsb::ops::process::signal(&pids, &named);
            // The OUTCOME, not the request. A signal is delivered
            // asynchronously, so the process is given a moment and then asked
            // whether it is still there — reporting "sent" as "stopped" is
            // what made this look like it did nothing.
            std::thread::sleep(std::time::Duration::from_millis(300));
            sent.into_iter()
                .map(|one| {
                    let alive = wsb::ops::process::is_alive(one.pid);
                    (one, alive)
                })
                .collect::<Vec<_>>()
        })
        .await
        .map_err(|error| {
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(serde_json::json!({ "error": format!("signalling failed: {error}") })),
            )
        })?;

        for (one, alive) in results {
            match one.outcome {
                wsb::ops::process::SignalOutcome::Sent if !alive => stopped.push(one.pid),
                // Delivered and still running: a process that catches the
                // signal and takes its time, or one that ignores it. Said
                // plainly rather than counted as a success.
                wsb::ops::process::SignalOutcome::Sent => refused.push(Refusal {
                    pid: one.pid,
                    reason: format!(
                        "{signal} was delivered and it is still running — send KILL to stop it \
                         outright"
                    ),
                }),
                wsb::ops::process::SignalOutcome::Gone => stopped.push(one.pid),
                wsb::ops::process::SignalOutcome::NotPermitted => refused.push(Refusal {
                    pid: one.pid,
                    reason: "another user owns this process, so this daemon may not stop it"
                        .to_string(),
                }),
                wsb::ops::process::SignalOutcome::Failed(said) => refused.push(Refusal {
                    pid: one.pid,
                    reason: said,
                }),
            }
        }
    }

    Ok(Json(KillOutcome {
        signalled: stopped,
        refused,
    }))
}

#[cfg(test)]
#[path = "op_processes_tests.rs"]
mod op_processes_tests;