haematite-cli 0.21.7

use super::{Config, Launch, Phase, Shared, io};
use std::fs::OpenOptions;
use std::os::unix::fs::OpenOptionsExt;
use std::process::{Child, Command, Stdio};
use std::sync::Arc;
use std::time::Instant;

pub(super) fn run_child(
    shared: &Arc<Shared>,
    index: usize,
    after: Option<usize>,
    config: &Config,
) -> Result<(), String> {
    if !wait_predecessor(shared, index, after, config)? {
        return Ok(());
    }
    let mut failures = 0;
    loop {
        let attempt = match launch(shared, index, config) {
            Ok(Some((child, launch))) => wait_child(shared, index, launch, child, config),
            Ok(None) => return Ok(()),
            Err(error) => Err(error),
        };
        let (ready, status) = match attempt {
            Ok(attempt) => attempt,
            Err(error) => {
                eprintln!("{error}");
                (false, error)
            }
        };
        let mut state = shared
            .state
            .lock()
            .map_err(|e| format!("door_children_state_failed: {e}"))?;
        if state.stopped {
            return Ok(());
        }
        state.entries[index].phase = Phase::Exited;
        drop(state.entries[index].stdin.take());
        drop(state);
        shared.changed.notify_all();
        if ready {
            failures = 0;
        } else {
            failures += 1;
        }
        if failures == config.give_up_after {
            shared.given_up(
                index,
                config,
                &format!("after {failures} exits before ready; last status {status}"),
            );
            return Ok(());
        }
    }
}

fn wait_predecessor(
    shared: &Shared,
    index: usize,
    after: Option<usize>,
    config: &Config,
) -> Result<bool, String> {
    let mut state = shared
        .state
        .lock()
        .map_err(|e| format!("door_children_state_failed: {e}"))?;
    if let Some(name) = config.after.as_deref()
        && after.is_some_and(|p| !state.entries[p].seen_ready)
    {
        eprintln!("door_child_waiting: {} after {name}", config.name);
    }
    while !state.stopped {
        if let Some(name) = config.after.as_deref()
            && after.is_some_and(|p| {
                state.entries[p].phase == Phase::GivenUp
                    && !state.entries[p].seen_ready
                    && state.entries[p].pending_ready == 0
            })
        {
            drop(state);
            shared.given_up(
                index,
                config,
                &format!("waiting after given-up child {name}"),
            );
            return Ok(false);
        }
        if (index == 0 || state.entries[index - 1].phase != Phase::Waiting)
            && after.is_none_or(|p| state.entries[p].seen_ready)
        {
            return Ok(true);
        }
        state = shared
            .changed
            .wait(state)
            .map_err(|e| format!("door_children_state_failed: {e}"))?;
    }
    drop(state);
    Ok(false)
}

fn launch(
    shared: &Arc<Shared>,
    index: usize,
    config: &Config,
) -> Result<Option<(Child, Launch)>, String> {
    let mut state = shared
        .state
        .lock()
        .map_err(|e| format!("door_children_state_failed: {e}"))?;
    if state.stopped {
        return Ok(None);
    }
    super::super::check::child(config)?;
    let entry = &mut state.entries[index];
    entry.generation = entry.generation.checked_add(1).ok_or_else(|| {
        format!(
            "door_child_start_failed: {}: launch generation exhausted",
            config.name
        )
    })?;
    let generation = entry.generation;
    entry.phase = Phase::NotReady;
    let log = OpenOptions::new()
        .create(true)
        .append(true)
        .mode(0o600)
        .open(&config.log)
        .map_err(|error| format!("door_child_log_failed: {}: {error}", config.name))?;
    let program = std::fs::canonicalize(&config.program)
        .map_err(|error| format!("door_child_start_failed: {}: {error}", config.name))?;
    let mut command = Command::new(&config.program);
    command.args(&config.args);
    if program.file_name().is_some_and(|name| name == "haematite")
        && !config.args.iter().any(|arg| arg == "--exit-on-stdin-close")
    {
        command.arg("--exit-on-stdin-close");
    }
    command
        .stdin(Stdio::piped())
        .stderr(Stdio::piped())
        .stdout(Stdio::inherit());
    let mut child = crate::process_spawn::locked(|| command.spawn())
        .map_err(|error| format!("door_child_start_failed: {}: {error}", config.name))?;
    let launch = Launch {
        generation,
        started: Instant::now(),
    };
    entry.stdin = child.stdin.take();
    #[cfg(test)]
    {
        entry.starts += 1;
    }
    let failed = Arc::clone(&entry.log_failed);
    let pipe = child.stderr.take();
    drop(state);
    shared.changed.notify_all();
    let Some(pipe) = pipe else {
        close_stdin(shared, index, generation)?;
        child
            .wait()
            .map_err(|error| format!("door_child_wait_failed: {}: {error}", config.name))?;
        return Err(format!(
            "door_child_stderr_failed: {}: pipe missing",
            config.name
        ));
    };
    let reading = Arc::clone(shared);
    let owned = config.clone();
    match std::thread::Builder::new()
        .name(format!("door-stderr-{}", config.name))
        .spawn(move || {
            if let Err(error) = io::read_lines(&reading, index, launch, &owned, pipe, log, failed) {
                eprintln!("{error}");
                if let Err(error) = close_stdin(&reading, index, generation) {
                    eprintln!("{error}");
                }
            }
        }) {
        // Descendants may hold the old pipe after the direct child exits.
        Ok(reader) => drop(reader),
        Err(error) => {
            eprintln!(
                "door_child_stderr_failed: {}: reader thread: {error}",
                config.name
            );
            close_stdin(shared, index, generation)?;
        }
    }
    Ok(Some((child, launch)))
}

fn close_stdin(shared: &Shared, index: usize, generation: u64) -> Result<(), String> {
    let mut state = shared
        .state
        .lock()
        .map_err(|e| format!("door_children_state_failed: {e}"))?;
    if state.entries[index].generation == generation
        && matches!(state.entries[index].phase, Phase::NotReady | Phase::Ready)
    {
        state.entries[index].phase = Phase::NotReady;
        drop(state.entries[index].stdin.take());
    }
    drop(state);
    Ok(())
}

fn wait_child(
    shared: &Shared,
    index: usize,
    launch: Launch,
    mut child: Child,
    config: &Config,
) -> Result<(bool, String), String> {
    let status = child
        .wait()
        .map_err(|error| format!("door_child_wait_failed: {}: {error}", config.name))?;
    let uptime_ms = launch.started.elapsed().as_millis();
    // An inherited stderr writer can outlive this process. The wait result
    // decides readiness and restart now; the reader drains independently to EOF.
    let mut state = shared
        .state
        .lock()
        .map_err(|e| format!("door_children_state_failed: {e}"))?;
    let entry = &mut state.entries[index];
    let ready = entry.generation == launch.generation && entry.phase == Phase::Ready;
    entry.phase = Phase::Exited;
    drop(entry.stdin.take());
    drop(state);
    eprintln!(
        "door_child_exited: {} {status} uptime_ms={uptime_ms}",
        config.name
    );
    Ok((ready, status.to_string()))
}