wsb 0.291.8591

//! The workshop's one daemon, run the way the workshop is set up.
//!
//! wsb runs its daemon two ways: on demand, started by the first command that
//! needs one, or as a service a supervisor keeps (`wsb daemon install`).
//! Nothing reconciled the two. After a migration a `wsb daemon restart` found
//! the service stopped and started a daemon on demand; the `wsb daemon
//! install` after it started the service, whose daemon found the port taken,
//! refused, and was started again by systemd every five seconds for ten
//! minutes — while `wsb daemon status` said the service was running.
//!
//! So every command that starts, replaces or stops the daemon comes through
//! here, and ends with one daemon of this wsb's version: the service's when a
//! service is installed, one started on demand otherwise. A daemon in the way
//! is stopped and named — one the service is not keeping, or one of another
//! version — and the command waits until the service's own daemon answers
//! before it says anything worked. When it cannot, it fails, saying what the
//! supervisor is doing instead.

use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

use super::client::Client;
use super::workshop::WorkshopRendezvous;
use super::service::{self, Control, Record, Supervisor, System};

/// How long the service's daemon is given to answer once started: a cold
/// start opens the database and reads the workshop's sessions before it
/// publishes itself. A daemon that is failing is reported as soon as the
/// supervisor says so, not after this.
const START_WITHIN: Duration = Duration::from_secs(60);

/// How long a daemon told to stop is given to be gone.
const STOP_WITHIN: Duration = Duration::from_secs(20);

const POLL: Duration = Duration::from_millis(100);

/// A daemon serving the workshop.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Serving {
    pub pid: u32,
    pub version: String,
    /// The workbench's port, when this build can read what it published.
    pub port: Option<u16>,
}

impl Serving {
    fn named(&self) -> String {
        format!("wsb-daemon {} (pid {})", self.version, self.pid)
    }
}

/// The daemons of a workshop: which one serves it, and stopping one.
pub(crate) trait Daemons {
    fn serving(&mut self) -> Result<Option<Serving>, String>;
    /// Stop it, and return once it is gone.
    fn stop(&mut self, daemon: &Serving) -> Result<(), String>;
}

/// This workshop's daemons, found through the rendezvous file in its host
/// folder.
pub(crate) struct PublishedDaemons<'a> {
    pub state_dir: &'a Path,
}

impl Daemons for PublishedDaemons<'_> {
    fn serving(&mut self) -> Result<Option<Serving>, String> {
        if let Some(published) = WorkshopRendezvous::read(self.state_dir)? {
            if let Ok(info) = Client::from_rendezvous(&published)?.info() {
                return Ok(Some(Serving { pid: info.pid, version: info.version, port: Some(info.http_port) }));
            }
            // Too busy to answer is still serving.
            return Ok(super::launch::still_serving(&published).then(|| Serving {
                pid: published.pid,
                version: published.version.clone(),
                port: Some(published.http_port),
            }));
        }
        // Published by a build that writes another shape: found by its pid.
        Ok(WorkshopRendezvous::published_by(self.state_dir)?
            .filter(|published| crate::ops::process::is_alive(published.pid))
            .map(|published| Serving { pid: published.pid, version: published.version, port: None }))
    }

    fn stop(&mut self, daemon: &Serving) -> Result<(), String> {
        // Through its control API when this build can speak it, else by the
        // signal every daemon shuts down in order on.
        let asked = match WorkshopRendezvous::read(self.state_dir)? {
            Some(published) if published.pid == daemon.pid => {
                Client::from_rendezvous(&published).and_then(|client| client.shutdown()).is_ok()
            }
            _ => false,
        };
        if !asked {
            if let Some(failed) = crate::ops::process::signal(&[daemon.pid], "TERM")
                .into_iter()
                .find(|signalled| signalled.outcome.is_failure())
            {
                return Err(format!("cannot stop {}: {:?}", daemon.named(), failed.outcome));
            }
        }
        let deadline = Instant::now() + STOP_WITHIN;
        while crate::ops::process::is_alive(daemon.pid) {
            if Instant::now() >= deadline {
                return Err(format!(
                    "{} was asked to stop and is still running {}s later; end it with `kill {}` and run this again",
                    daemon.named(),
                    STOP_WITHIN.as_secs(),
                    daemon.pid
                ));
            }
            std::thread::sleep(POLL);
        }
        WorkshopRendezvous::remove_if_published_by(self.state_dir, daemon.pid).map(|_| ())
    }
}

/// The service installed on this host.
#[derive(Debug, Clone)]
pub struct Installed {
    pub record: Record,
    pub supervisor: Supervisor,
    pub definition: PathBuf,
}

impl Installed {
    /// Whether the definition on disk is not what the record renders, or is
    /// not there: what a supervisor would start is not what was installed.
    fn stale(&self) -> bool {
        match std::fs::read_to_string(&self.definition) {
            Ok(on_disk) => service::drifted(self.supervisor, &self.record, Some(&on_disk)),
            Err(_) => true,
        }
    }
}

/// The service this workshop runs its daemon as, when it has one.
pub fn installed(state_dir: &Path) -> Result<Option<Installed>, String> {
    let Some(record) = service::record(state_dir)? else {
        return Ok(None);
    };
    let supervisor = service::supervisor()?;
    let definition = service::definition_path(supervisor)?;
    Ok(Some(Installed { record, supervisor, definition }))
}

/// What bringing the service's daemon up did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Brought {
    /// The daemon now serving, the service's.
    pub serving: Serving,
    /// Each daemon that was in its way, and what became of it.
    pub replaced: Vec<String>,
}

/// Make the service's daemon the one serving the workshop, at this wsb's
/// version.
///
/// A daemon serving that the service is not keeping, or of another version,
/// is stopped first; `restart` stops the service's own as well. `prepare`
/// then writes what the supervisor reads, and the service is started. Done
/// when the daemon the supervisor is keeping is the one answering; failed as
/// soon as the supervisor says it is failing, or when it has not answered in
/// time.
#[allow(clippy::too_many_arguments)]
pub(crate) fn bring_up_with<C: Control, D: Daemons>(
    control: &mut C,
    daemons: &mut D,
    supervisor: Supervisor,
    definition: &Path,
    version: &str,
    restart: bool,
    prepare: impl FnOnce(&mut C) -> Result<(), String>,
    within: Duration,
    poll: Duration,
) -> Result<Brought, String> {
    let mut replaced = Vec::new();
    let keeping = service::supervised_with(control, supervisor)?.keeping();
    if let Some(daemon) = daemons.serving()? {
        let owned = keeping == Some(daemon.pid);
        let current = daemon.version == version;
        if !owned || !current || restart {
            daemons.stop(&daemon)?;
            if !owned {
                replaced.push(format!("stopped {}, which the service was not keeping", daemon.named()));
            } else if !current {
                replaced.push(format!("stopped the service's {}, which is not this wsb's {version}", daemon.named()));
            }
        }
    }
    prepare(control)?;
    service::start_with(control, supervisor, definition)?;

    let deadline = Instant::now() + within;
    loop {
        let state = service::supervised_with(control, supervisor)?;
        if state.failing() {
            return Err(format!("the service is {}", state.failure()));
        }
        if let Some(daemon) = daemons.serving()? {
            match state.keeping() {
                Some(pid) if pid == daemon.pid && daemon.version == version => {
                    return Ok(Brought { serving: daemon, replaced });
                }
                Some(pid) if pid == daemon.pid => {
                    return Err(format!(
                        "the service started {}, and this is wsb {version}: its definition names another \
                         build. Install it again with this wsb: wsb daemon install",
                        daemon.named()
                    ));
                }
                // Started meanwhile by something else, which the service
                // cannot serve beside.
                _ => {
                    daemons.stop(&daemon)?;
                    replaced.push(format!("stopped {}, started while the service was starting", daemon.named()));
                }
            }
        }
        if Instant::now() >= deadline {
            return Err(format!(
                "the service's daemon has not answered in {}s; {} says it is {}",
                within.as_secs(),
                supervisor.describe(),
                state.failure_or_phase()
            ));
        }
        std::thread::sleep(poll);
    }
}

fn bring_up_here(
    state_dir: &Path,
    installed: &Installed,
    restart: bool,
    prepare: impl FnOnce(&mut System) -> Result<(), String>,
) -> Result<Brought, String> {
    bring_up_with(
        &mut System,
        &mut PublishedDaemons { state_dir },
        installed.supervisor,
        &installed.definition,
        crate::get_version(),
        restart,
        prepare,
        START_WITHIN,
        POLL,
    )
    .map_err(|error| with_last_error(state_dir, error))
}

/// A failure, with the last error the daemon logged: what an operator needs
/// to see next, and otherwise has to go and find.
fn with_last_error(state_dir: &Path, error: String) -> String {
    match last_error(state_dir) {
        Some(logged) => format!("{error}\n  its log says: {logged}"),
        None => error,
    }
}

/// The last error line in the daemon's log.
pub fn last_error(state_dir: &Path) -> Option<String> {
    let text = std::fs::read_to_string(super::log_path(state_dir)).ok()?;
    let tail = &text[text.len().saturating_sub(64 * 1024)..];
    tail.lines().rev().find(|line| line.contains("Error:")).map(|line| line.trim().to_string())
}

/// Put the service on this host as `record` says, and bring its daemon up.
pub fn install(state_dir: &Path, record: &Record) -> Result<Brought, String> {
    let installed = Installed {
        record: record.clone(),
        supervisor: service::supervisor()?,
        definition: service::definition_path(service::supervisor()?)?,
    };
    bring_up_here(state_dir, &installed, false, |control| {
        service::apply_with(control, installed.supervisor, record, &installed.definition, service::BOOTOUT_TIMEOUT)?;
        service::write_record(state_dir, record)
    })
}

/// Start the service at login, and bring its daemon up now.
pub fn enable(state_dir: &Path) -> Result<(Record, Brought), String> {
    let Some(mut installed) = installed(state_dir)? else {
        return Err("no service is installed on this host, so there is nothing to enable.\n  \
                    `wsb daemon install` puts one on it."
            .to_string());
    };
    installed.record.autostart = true;
    let record = installed.record.clone();
    let brought = bring_up_here(state_dir, &installed, false, |control| {
        service::apply_with(control, installed.supervisor, &record, &installed.definition, service::BOOTOUT_TIMEOUT)?;
        service::write_record(state_dir, &record)
    })?;
    Ok((record, brought))
}

/// Bring the installed service's daemon up, the service's own stopped and
/// started again too when `restart`. A definition that is not what the record
/// renders is written again first.
pub fn bring_up(state_dir: &Path, installed: &Installed, restart: bool) -> Result<Brought, String> {
    let stale = installed.stale();
    bring_up_here(state_dir, installed, restart, |control| {
        if stale {
            service::apply_with(
                control,
                installed.supervisor,
                &installed.record,
                &installed.definition,
                service::BOOTOUT_TIMEOUT,
            )
        } else {
            Ok(())
        }
    })
}

/// Stop the daemon: the service, so its supervisor does not start it again,
/// and any daemon serving besides. What was stopped.
pub fn stop(state_dir: &Path) -> Result<Vec<String>, String> {
    // Whether a service is installed is whether its record is there; stopping
    // it needs only the supervisor, never what the record says. A record an
    // older wsb wrote is `wsb migrate`'s to convert, and migrate needs the
    // daemon stopped first: reading the record here made each wait on the
    // other.
    stop_in(state_dir, &mut System, &mut PublishedDaemons { state_dir }, service::supervisor)
}

/// [`stop`], with the supervisor, the way to find which one it is, and the
/// daemons given.
pub(crate) fn stop_in(
    state_dir: &Path,
    control: &mut impl service::Control,
    daemons: &mut impl Daemons,
    supervisor_of_this_host: impl FnOnce() -> Result<Supervisor, String>,
) -> Result<Vec<String>, String> {
    let supervisor = match service::record_path(state_dir).exists() {
        true => Some(supervisor_of_this_host()?),
        false => None,
    };
    let mut stopped = Vec::new();
    if let Some(supervisor) = supervisor {
        let state = service::supervised_with(control, supervisor)?;
        if state.phase != service::Phase::Stopped {
            let was = daemons.serving()?.filter(|daemon| state.keeping() == Some(daemon.pid));
            service::stop_with(control, supervisor)?;
            stopped.push(match was {
                Some(daemon) => format!("stopped the service's {}", daemon.named()),
                None => format!("stopped the service ({})", state.failure_or_phase()),
            });
        }
    }
    if let Some(daemon) = daemons.serving()? {
        daemons.stop(&daemon)?;
        stopped.push(format!("stopped {}", daemon.named()));
    }
    Ok(stopped)
}

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