zc2 0.0.29

//! What the node's own supervisor says about the worker process.
//!
//! The image a node runs (`zakuroai/compute`, see the `zakuro-image` repo)
//! carries an s6 supervisor with a small HTTP API. The broker runs as an s6
//! service in that same container, so it can ask about its neighbours over
//! loopback -- no mesh, no inbound port, nothing exposed.
//!
//! Reported alongside the worker sync the broker already sends, deliberately:
//! the dashboard pulling this instead would mean the platform making outbound
//! calls to user machines, where one unreachable node stalls a page render.
//!
//! Only three fields leave the container. `/api/services/{name}/details` also
//! returns `run_script` and `stdout`, which are the script's source and the
//! service's output -- neither is health, both could carry anything, and
//! forwarding them would turn a status column into an exfiltration path.
//! `exit_code` is NOT available here: it belongs to `POST /api/terminal/exec`,
//! a different endpoint entirely, which this never calls.

use std::time::Duration;

/// The s6 service the Zakuro worker runs as, per the compute image's
/// `/etc/s6-overlay/s6-rc.d/zakuro_worker`.
pub const WORKER_SERVICE: &str = "zakuro_worker";

/// Where the supervisor's API listens inside the image (`PORT=8080`).
const DEFAULT_S6_URL: &str = "http://127.0.0.1:8080";

/// Short: this runs inside a sync the broker must finish promptly, and a
/// supervisor that cannot answer in a moment is not worth waiting for.
const TIMEOUT: Duration = Duration::from_millis(500);

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WorkerHealth {
    /// s6's own word for it: "running", "down", …
    pub status: String,
    /// How many times the supervisor has restarted it. The number that shows
    /// a worker flapping rather than merely being up right now.
    pub restart_count: Option<i64>,
    /// As s6 reports it, e.g. "120s". A string on purpose -- the supervisor's
    /// format is its own, and reformatting it here would invent precision.
    pub uptime: Option<String>,
}

/// Pull the three fields out of a details response.
///
/// Returns `None` for anything unparseable rather than guessing: this is a
/// nicety attached to a sync that must not fail because a supervisor changed
/// its JSON.
pub fn parse_details(body: &str) -> Option<WorkerHealth> {
    let v: serde_json::Value = serde_json::from_str(body).ok()?;
    let status = v.get("status")?.as_str()?.to_string();
    if status.is_empty() {
        return None;
    }
    Some(WorkerHealth {
        status,
        // s6manager reports pid as a string, so accept a number or a numeric
        // string for the count too rather than trusting one shape.
        restart_count: v.get("restart_count").and_then(|r| {
            r.as_i64()
                .or_else(|| r.as_str().and_then(|s| s.parse().ok()))
        }),
        uptime: v
            .get("uptime")
            .and_then(|u| u.as_str())
            .filter(|s| !s.is_empty())
            .map(str::to_string),
    })
}

fn s6_base() -> String {
    std::env::var("ZAKURO_S6_URL").unwrap_or_else(|_| DEFAULT_S6_URL.to_string())
}

/// Ask the local supervisor about the worker service.
///
/// `None` whenever there is no answer, which is the common case rather than an
/// error: a node run outside the image has no supervisor to ask, and the sync
/// carries on reporting everything else.
pub fn worker_health() -> Option<WorkerHealth> {
    let url = format!(
        "{}/api/services/{}/details",
        s6_base().trim_end_matches('/'),
        WORKER_SERVICE
    );
    let agent = ureq::Agent::new_with_config(
        ureq::Agent::config_builder()
            .timeout_connect(Some(TIMEOUT))
            .timeout_recv_response(Some(TIMEOUT))
            .build(),
    );
    let body = agent
        .get(&url)
        .call()
        .ok()?
        .body_mut()
        .read_to_string()
        .ok()?;
    parse_details(&body)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parses_a_healthy_service() {
        let h = parse_details(
            r#"{"name":"zakuro_worker","status":"running","uptime":"120s",
                "pid":"52","restart_count":0}"#,
        )
        .unwrap();
        assert_eq!(h.status, "running");
        assert_eq!(h.restart_count, Some(0));
        assert_eq!(h.uptime.as_deref(), Some("120s"));
    }

    #[test]
    fn parses_a_flapping_service() {
        let h = parse_details(r#"{"status":"down","restart_count":17}"#).unwrap();
        assert_eq!(h.status, "down");
        assert_eq!(h.restart_count, Some(17));
    }

    #[test]
    fn accepts_a_numeric_string_restart_count() {
        // pid comes back as a string in this API; do not assume the count
        // is always a number.
        let h = parse_details(r#"{"status":"running","restart_count":"3"}"#).unwrap();
        assert_eq!(h.restart_count, Some(3));
    }

    #[test]
    fn missing_optional_fields_are_none_not_zero() {
        // A count of 0 means the supervisor said zero. Absent means it did
        // not say, and reporting 0 for that would claim a healthy history.
        let h = parse_details(r#"{"status":"running"}"#).unwrap();
        assert_eq!(h.restart_count, None);
        assert_eq!(h.uptime, None);
    }

    #[test]
    fn no_status_is_no_health() {
        assert!(parse_details(r#"{"uptime":"5s"}"#).is_none());
        assert!(parse_details(r#"{"status":""}"#).is_none());
    }

    #[test]
    fn junk_does_not_panic_or_guess() {
        for body in ["", "not json", "[]", "null", r#"{"status":42}"#] {
            assert!(parse_details(body).is_none(), "body: {body:?}");
        }
    }

    #[test]
    fn nothing_but_the_three_fields_is_carried() {
        // run_script and stdout are the script's source and the service's
        // output. Neither is health, and forwarding them would make a status
        // column an exfiltration path.
        // r##"..."## because the payload itself contains `"#`, which would
        // close a single-hash raw string.
        let h = parse_details(
            r##"{"status":"running","run_script":"#!/bin/sh\nexec secret",
                "stdout":"KEY=hunter2","restart_count":1}"##,
        )
        .unwrap();
        assert_eq!(h.status, "running");
        assert_eq!(h.restart_count, Some(1));
        assert_eq!(h.uptime, None);
    }
}