qex 0.10.0

//! This module finds the monitor scripts that wait for a proxy.
//!
//! A monitor waits for evidence of the work: a pattern in the process list, a
//! line in a log file, a file that appears. Evidence stops arriving when the
//! work stops, and the monitor cannot see that. Such a monitor then sleeps for
//! ever. Four of them on one machine slept for 95 hours between them.
//!
//! # Why this command can do what a shell command cannot
//!
//! The usual way to look for these monitors is `pgrep -f pgrep`, and that
//! command matches itself: its own command line holds the letters that it looks
//! for. A user of that method finds the search and reads it as a monitor.
//!
//! This command knows its own process id, its own process group and its own
//! ancestors, so it removes them before it reports anything. It cannot find
//! itself.
//!
//! # What this command cannot see
//!
//! It reads the command line of each process. A monitor that a user wrote
//! INSIDE A SCRIPT FILE gives the name of that script in its command line, and
//! not the loop, so this command cannot classify it. The report says so.

use serde::Serialize;

/// One monitor that this command found.
#[derive(Debug, Clone, Serialize)]
pub struct Watcher {
    pub pid: i32,
    /// The time that this process has operated, in seconds.
    pub age_secs: u64,
    pub command: String,
    /// The kind of proxy that this monitor waits for.
    pub proxy: &'static str,
    /// What a reader should do about it.
    pub advice: &'static str,
}

/// Tests one command line, and gives the kind of proxy that it waits for.
///
/// This function looks for a loop that sleeps. A command that sleeps once is
/// not a monitor.
pub fn classify(command: &str) -> Option<(&'static str, &'static str)> {
    let lower = command.to_ascii_lowercase();

    // A monitor sleeps in a loop. Without a sleep, a command that holds these
    // words is doing its work and not waiting for a proxy.
    let sleeps = lower.contains("sleep");
    let loops = lower.contains("while ") || lower.contains("until ") || lower.contains("for ");
    if !(sleeps && loops) {
        return None;
    }

    let reads_processes = lower.contains("pgrep")
        || lower.contains("pidof")
        || lower.contains("ps -ef")
        || lower.contains("ps aux")
        || lower.contains("ps -a");

    // A COUNT of the processes that match, and not a test of one process.
    //
    // This one is the worst of the group, and a careful author writes it. It
    // has no pattern fault: the monitor waits until NOTHING matches. On a
    // machine that two agents share, that condition is not satisfiable, because
    // the work of the other agent holds the count above zero for ever. The work
    // of this author is already complete, and the monitor cannot see the
    // difference.
    //
    // A user found one of these after 63 hours. It counted a program over
    // `ssh` on a machine that another agent also used, and it opened about 750
    // connections to that machine while it waited.
    let counts = lower.contains("grep -c") || lower.contains("wc -l") || lower.contains("--count");

    if reads_processes && counts {
        return Some((
            "a count of the processes that match, which another user holds above zero",
            "This monitor waits until NOTHING matches, and it has no pattern fault. On a \
             machine that two agents share, the work of the other agent keeps the count above \
             zero for ever, and your own work is already complete. Wait for YOUR job instead: \
             `qex wait <id>` reads one process, and that process belongs to you.",
        ));
    }

    if reads_processes && lower.contains("ssh") {
        return Some((
            "a pattern in the process list of another machine",
            "That pattern also matches the work of the other users of that machine, and each \
             test opens a connection. Start the task with `qex submit` on that machine, and \
             wait for its id.",
        ));
    }

    if reads_processes {
        return Some((
            "a pattern in the process list",
            "This monitor can match its own command line. Use `qex submit` and \
             `qex status <id> --wait`.",
        ));
    }

    if lower.contains("grep") || lower.contains("tail ") {
        return Some((
            "a line in a log file",
            "The line never arrives when something stops the task that writes it. \
             Use `qex submit` and `qex status <id> --wait`.",
        ));
    }

    if lower.contains("test -f") || lower.contains("[ -f") || lower.contains("[ -e") {
        return Some((
            "a file that appears",
            "The file never appears when something stops the task that writes it. \
             Use `qex submit` and `qex status <id> --wait`.",
        ));
    }

    Some((
        "an unknown condition",
        "This loop sleeps and tests a condition. Use `qex submit` and \
         `qex status <id> --wait`, which waits for the process itself.",
    ))
}

/// Finds the monitors that operate now.
///
/// The result never holds this process, its process group, or any process that
/// started it. A command that looks for this fault must not find itself.
#[cfg(target_os = "linux")]
pub fn find() -> Vec<Watcher> {
    let mut out = Vec::new();

    let me = std::process::id() as i32;
    let ancestors = ancestors_of(me);

    let Ok(entries) = std::fs::read_dir("/proc") else {
        return out;
    };

    let uptime = read_uptime();
    let ticks = unsafe { libc::sysconf(libc::_SC_CLK_TCK) } as f64;

    for entry in entries.flatten() {
        let name = entry.file_name();
        let Some(name) = name.to_str() else { continue };
        let Ok(pid) = name.parse::<i32>() else {
            continue;
        };

        // Remove this process and the processes that started it. This step is
        // the reason that this command can look for a fault that a shell
        // command cannot look for without finding itself: the shell that runs
        // `qex watchers` holds those letters in its own command line.
        //
        // Remove these processes only. A monitor that a user started from the
        // same shell is a true result, and it is the usual case.
        if pid == me || ancestors.contains(&pid) {
            continue;
        }

        let Ok(raw) = std::fs::read(entry.path().join("cmdline")) else {
            continue;
        };
        if raw.is_empty() {
            continue;
        }
        // The parts of a command line are separated by a zero byte.
        let command = String::from_utf8_lossy(&raw)
            .replace('\0', " ")
            .trim()
            .to_string();

        // A command that runs qex itself is not a monitor.
        if command.contains("qex watchers") {
            continue;
        }

        let Some((proxy, advice)) = classify(&command) else {
            continue;
        };

        out.push(Watcher {
            pid,
            age_secs: process_age(&entry.path(), uptime, ticks),
            command,
            proxy,
            advice,
        });
    }

    // The oldest first. A monitor that has slept for a day is the clearest
    // fault, and the most useful line for a reader.
    out.sort_by_key(|w| std::cmp::Reverse(w.age_secs));
    out
}

#[cfg(not(target_os = "linux"))]
pub fn find() -> Vec<Watcher> {
    // macOS has no `/proc`. Read the process list with `ps`, and remove this
    // process and its group in the same way.
    let mut out = Vec::new();
    let me = std::process::id() as i32;
    let my_group = unsafe { libc::getpgid(0) };

    let Ok(result) = std::process::Command::new("ps")
        .args(["-A", "-o", "pid=,pgid=,etime=,command="])
        .output()
    else {
        return out;
    };

    for line in String::from_utf8_lossy(&result.stdout).lines() {
        let mut parts = line.trim().splitn(4, char::is_whitespace);
        let (Some(pid), Some(group), Some(elapsed), Some(command)) =
            (parts.next(), parts.next(), parts.next(), parts.next())
        else {
            continue;
        };
        let Ok(pid) = pid.parse::<i32>() else {
            continue;
        };
        if pid == me || group.parse::<i32>() == Ok(my_group) {
            continue;
        }
        let Some((proxy, advice)) = classify(command) else {
            continue;
        };
        out.push(Watcher {
            pid,
            age_secs: parse_elapsed(elapsed),
            command: command.to_string(),
            proxy,
            advice,
        });
    }

    out.sort_by_key(|w| std::cmp::Reverse(w.age_secs));
    out
}

#[cfg(not(target_os = "linux"))]
fn parse_elapsed(text: &str) -> u64 {
    // `ps` gives [[DD-]HH:]MM:SS.
    let (days, rest) = match text.split_once('-') {
        Some((d, r)) => (d.parse::<u64>().unwrap_or(0), r),
        None => (0, text),
    };
    let mut seconds = 0u64;
    for part in rest.split(':') {
        seconds = seconds * 60 + part.parse::<u64>().unwrap_or(0);
    }
    days * 86400 + seconds
}

/// Gives the process ids that started this process.
#[cfg(target_os = "linux")]
fn ancestors_of(mut pid: i32) -> Vec<i32> {
    let mut out = Vec::new();
    // A limit, so a strange process table cannot make an endless loop.
    for _ in 0..64 {
        let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat")) else {
            break;
        };
        let Some(rest) = stat.rsplit_once(") ") else {
            break;
        };
        let fields: Vec<&str> = rest.1.split_whitespace().collect();
        if fields.len() < 2 {
            break;
        }
        let Ok(parent) = fields[1].parse::<i32>() else {
            break;
        };
        if parent <= 1 {
            break;
        }
        out.push(parent);
        pid = parent;
    }
    out
}

#[cfg(target_os = "linux")]
fn read_uptime() -> f64 {
    std::fs::read_to_string("/proc/uptime")
        .ok()
        .and_then(|t| t.split_whitespace().next().map(|s| s.to_string()))
        .and_then(|s| s.parse::<f64>().ok())
        .unwrap_or(0.0)
}

#[cfg(target_os = "linux")]
fn process_age(dir: &std::path::Path, uptime: f64, ticks: f64) -> u64 {
    let Ok(stat) = std::fs::read_to_string(dir.join("stat")) else {
        return 0;
    };
    let Some(rest) = stat.rsplit_once(") ") else {
        return 0;
    };
    let fields: Vec<&str> = rest.1.split_whitespace().collect();
    // After the command, field 20 is the time when the process started.
    if fields.len() < 20 {
        return 0;
    }
    let started: f64 = fields[19].parse().unwrap_or(0.0);
    (uptime - started / ticks).max(0.0) as u64
}

/// Writes the monitors that this command found.
pub fn report(json: bool) -> anyhow::Result<i32> {
    let found = find();

    if json {
        println!("{}", serde_json::to_string_pretty(&found)?);
        return Ok(if found.is_empty() { 0 } else { 1 });
    }

    if found.is_empty() {
        println!("no monitor script waits for a proxy on this machine.");
        return Ok(0);
    }

    let total: u64 = found.iter().map(|w| w.age_secs).sum();
    println!(
        "{} monitor script(s) wait for a proxy. Together they have waited {}.",
        found.len(),
        crate::units::format_duration(std::time::Duration::from_secs(total))
    );
    println!();

    for w in &found {
        println!(
            "{}  pid {}  waiting {}",
            crate::style::warning("MONITOR"),
            w.pid,
            crate::units::format_duration(std::time::Duration::from_secs(w.age_secs))
        );
        println!("  waits for: {}", w.proxy);
        println!("  command:   {:.120}", w.command);
        println!("  {}", w.advice);
        println!();
    }

    println!(
        "{}",
        crate::style::faint(
            "This command removes its own process and the processes that started it \
             before it reports anything, so it never finds itself.\nStop one with \
             `kill <pid>`. A kill of the shell does not stop the processes that the shell \
             started; those go to the init process and continue.\nA monitor INSIDE A SCRIPT \
             FILE gives the name of the script in its command line, and not the loop, so this \
             command cannot see it. Read the script."
        )
    );
    Ok(1)
}

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

    /// The three monitors that a user measured on one machine.
    #[test]
    fn the_real_monitors_are_recognised() {
        let (proxy, _) = classify("bash -c while pgrep -f solve.py; do sleep 60; done").unwrap();
        assert_eq!(proxy, "a pattern in the process list");

        let (proxy, _) =
            classify("bash -c until grep -q 'DONE' run.log; do sleep 60; done").unwrap();
        assert_eq!(proxy, "a line in a log file");

        let (proxy, _) =
            classify("sh -c until [ -f /tmp/done.marker ]; do sleep 30; done").unwrap();
        assert_eq!(proxy, "a file that appears");
    }

    /// The fourth kind, which a user measured at 63 hours.
    ///
    /// This monitor has NO PATTERN FAULT. It waits until nothing matches, and
    /// on a machine that two agents share that condition is not satisfiable:
    /// the work of the other agent holds the count above zero for ever, while
    /// the work of the author is already complete. A careful author writes this
    /// one, which is the reason that it has its own words.
    #[test]
    fn a_count_that_another_user_holds_above_zero_is_recognised() {
        let (proxy, advice) = classify(
            "bash -c while true; do M=$(ps -Ao args | grep -c solver); \
             K=$(ssh host ps -Ao args | grep -c solver); sleep 300; done",
        )
        .unwrap();
        assert!(
            proxy.contains("count"),
            "a count of the matches is its own fault: {proxy}"
        );
        assert!(
            advice.contains("two agents share"),
            "the advice must name the cause: {advice}"
        );

        // A pattern on another machine, with no count, is its own kind as well.
        let (proxy, _) =
            classify("sh -c until ssh host pgrep -f train.py; do sleep 60; done").unwrap();
        assert!(proxy.contains("another machine"), "got: {proxy}");

        // The ordinary self-match keeps its own words.
        let (proxy, _) = classify("bash -c while pgrep -f solve.py; do sleep 60; done").unwrap();
        assert_eq!(proxy, "a pattern in the process list");
    }

    /// A command that does its work must not look like a monitor.
    #[test]
    fn ordinary_commands_are_not_monitors() {
        assert!(classify("cargo test --release").is_none());
        assert!(
            classify("grep -r pattern src/").is_none(),
            "a search is not a loop"
        );
        assert!(classify("sleep 60").is_none(), "one sleep is not a loop");
        assert!(
            classify("pgrep -f something").is_none(),
            "one search of the process list is not a monitor"
        );
        assert!(
            classify("python3 train.py --epochs 50").is_none(),
            "a long task is not a monitor"
        );
    }

    /// A loop that sleeps and tests something else is still a monitor.
    #[test]
    fn an_unknown_condition_is_still_a_monitor() {
        let (proxy, advice) =
            classify("bash -c while ! curl -sf localhost:8080; do sleep 5; done").unwrap();
        assert_eq!(proxy, "an unknown condition");
        assert!(advice.contains("qex"));
    }

    /// This command must never report itself.
    ///
    /// A user hunted for this fault with `pgrep -f pgrep`, and the search
    /// matched its own command line. That was the fourth occurrence of the
    /// fault in one day, inside the hunt for it.
    #[test]
    fn the_search_never_finds_itself() {
        let me = std::process::id() as i32;
        let found = find();
        assert!(
            !found.iter().any(|w| w.pid == me),
            "the search reported itself"
        );

        // It must not report the processes that started it either. The shell
        // that runs this test holds the words of the command in its own line.
        //
        // The list of the ancestors comes from `/proc`, which is Linux only.
        // The macOS code removes the process group instead, and the test above
        // covers the part that both systems share.
        #[cfg(target_os = "linux")]
        for parent in ancestors_of(me) {
            assert!(
                !found.iter().any(|w| w.pid == parent),
                "the search reported a process that started it"
            );
        }
    }
}