aion-cli 0.31.0

//! Handing one Aion home over from the server running it to a fresh one.
//!
//! Two verbs need exactly this: `aion server restart` (same binary, new
//! process) and `aion update` (a NEWLY INSTALLED binary, new process). They
//! differ in which executable starts and in what they had to do first; the
//! handover itself — signal, wait for the predecessor to visibly let go, start
//! the successor, narrate both until the new one serves — is one procedure,
//! written once here.
//!
//! # Why the successor starts before the predecessor is gone
//!
//! The successor waits out the store's writer lock while the predecessor
//! drains. That overlap is the shortest gap this machine can offer, and the
//! narration names it: a successor sitting in `writer-lock-wait` looks exactly
//! like a wedge to anyone who has not been told what it is waiting for.
//!
//! # The race this closes, from both ends
//!
//! A birth claim that finds a live SERVING incarnation refuses itself as a
//! colliding sibling; only a DRAINING one is succeeded. So the server flips
//! its record to `draining` in the signal path BEFORE the drain runs, and this
//! module waits for that state (or for the process to be gone) before it
//! starts anything. Both halves are needed; either alone leaves a window.

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

use aion_server::control::{
    IncarnationState, PidRecord, StopRefusal, TerminationSignal, incarnation, send_termination,
};

use crate::boot_narration::{BootNarrator, BootObservation};
use crate::console::PROBE_STEP_TIMEOUT;

/// How often the handover re-reads the record and the process table. Derived
/// from the pid record's one-second stage resolution: polling faster cannot
/// see anything new.
const POLL_INTERVAL: Duration = Duration::from_millis(250);

/// A handover that did not happen, or did not complete, and what to tell the
/// operator about it.
pub struct HandoverRefusal {
    /// Process exit code.
    pub code: u8,
    /// Plain-language report for stderr.
    pub message: String,
}

impl HandoverRefusal {
    /// Nothing was signalled and nothing was started.
    #[must_use]
    pub fn refused(message: String) -> Self {
        Self { code: 2, message }
    }

    /// The handover was attempted and did not complete.
    #[must_use]
    pub fn incomplete(message: String) -> Self {
        Self { code: 1, message }
    }
}

/// The running server a handover replaces: a live, verified incarnation that
/// is SERVING or already DRAINING.
///
/// A BOOTING incarnation is refused. Replacing a boot means killing it
/// mid-store-open and starting the same work again from the beginning, which
/// is not a restart but a slower start; whoever wants that can say so with
/// `aion server stop`.
///
/// # Errors
///
/// Returns [`HandoverRefusal::refused`] when the home has no record, its
/// record names a dead or recycled pid, the record cannot be read, or the
/// recorded incarnation is still booting.
pub fn resolve_predecessor(home: &Path, verb: &str) -> Result<PidRecord, HandoverRefusal> {
    match crate::boot_narration::read_observation(home) {
        BootObservation::Live(record) => match record.state {
            IncarnationState::Serving | IncarnationState::Draining => Ok(*record),
            IncarnationState::Booting => Err(HandoverRefusal::refused(format!(
                "server {} is still BOOTING ({}); there is nothing serving to {verb}. \
                 Watch it with `aion server status`; if you must replace it, stop it \
                 with `aion server stop` and start a new one",
                record.pid,
                record
                    .stage_line()
                    .unwrap_or_else(|| "no stage reported yet".to_owned())
            ))),
        },
        BootObservation::NoRecord => Err(HandoverRefusal::refused(format!(
            "no server has claimed this home, so there is nothing to {verb}. Check with \
             `aion server status`; start one with `aion`"
        ))),
        BootObservation::NotLive(record) => Err(HandoverRefusal::refused(format!(
            "this home's record names pid {}, which is not a live incarnation of that \
             server (it died, or its pid was recycled). Nothing was signalled. Check \
             with `aion server status`; start a server with `aion`",
            record.pid
        ))),
        BootObservation::Unreadable(error) => Err(HandoverRefusal::refused(format!(
            "this home's pid record could not be read ({error}); nothing was signalled"
        ))),
    }
}

/// Signal the predecessor and wait until it has visibly let go: its record
/// reads `draining`, or its process is gone.
///
/// A predecessor already DRAINING is not signalled again — it is on its way
/// out, and a second termination signal is the operator's escalation to a
/// FORCED exit, which is not what a handover asks for.
///
/// # Errors
///
/// Returns [`HandoverRefusal::refused`] when the signal cannot be sent (so
/// nothing was started), and [`HandoverRefusal::incomplete`] when the
/// predecessor neither drains nor exits within `patience` — no successor is
/// started in that case, because one begun then would be refused as a
/// colliding sibling.
pub fn release_predecessor(
    home: &Path,
    predecessor: &PidRecord,
    patience: Duration,
) -> Result<(), HandoverRefusal> {
    if predecessor.state == IncarnationState::Draining {
        println!(
            "server {} (version {}) is already draining; starting its successor now",
            predecessor.pid, predecessor.version
        );
        return Ok(());
    }
    println!(
        "stopping server {} (version {}) — waiting up to {}s for it to begin draining",
        predecessor.pid,
        predecessor.version,
        patience.as_secs()
    );
    match send_termination(predecessor) {
        Ok(TerminationSignal::Sent) => {}
        Ok(TerminationSignal::AlreadyGone) => {
            println!(
                "server {} exited between the check and the signal; starting its successor",
                predecessor.pid
            );
            return Ok(());
        }
        Err(StopRefusal::UnsupportedPlatform) => {
            return Err(HandoverRefusal::refused(
                "a handover needs POSIX signals, which this platform does not have".to_owned(),
            ));
        }
        Err(refusal) => {
            return Err(HandoverRefusal::refused(format!(
                "nothing was started: {refusal}"
            )));
        }
    }
    wait_until_letting_go(home, predecessor, patience)
}

fn wait_until_letting_go(
    home: &Path,
    predecessor: &PidRecord,
    patience: Duration,
) -> Result<(), HandoverRefusal> {
    let started = Instant::now();
    loop {
        if !matches!(
            incarnation::probe(predecessor),
            incarnation::IncarnationProbe::Verified { .. }
        ) {
            println!("server {} has exited", predecessor.pid);
            return Ok(());
        }
        match crate::boot_narration::read_observation(home) {
            BootObservation::Live(record)
                if record.is_same_incarnation(predecessor)
                    && record.state == IncarnationState::Draining =>
            {
                println!("server {} is draining", predecessor.pid);
                return Ok(());
            }
            BootObservation::Unreadable(error) => {
                return Err(HandoverRefusal::incomplete(format!(
                    "server {} was signalled, but this home's pid record then became \
                     unreadable ({error}); no successor was started. Check `aion server \
                     status` before starting one",
                    predecessor.pid
                )));
            }
            BootObservation::Live(_) | BootObservation::NotLive(_) | BootObservation::NoRecord => {}
        }
        if started.elapsed() >= patience {
            return Err(HandoverRefusal::incomplete(format!(
                "server {} was signalled but has neither begun draining nor exited \
                 within {}s. No successor was started — one begun now would be refused \
                 as a colliding sibling. Watch it with `aion server status`, or force \
                 it with a second `aion server stop`",
                predecessor.pid,
                patience.as_secs()
            )));
        }
        std::thread::sleep(POLL_INTERVAL);
    }
}

/// Start `executable` as this home's server and narrate until it serves.
///
/// `predecessor` is the incarnation being replaced, when there is one; its
/// exit is reported as it happens (that is when the store's writer lock frees,
/// which is what the successor is usually waiting for), and it names the
/// "from" half of the final line.
///
/// Returns the successor's own record, so a caller with more to say than
/// "it serves" can say it about the incarnation that is actually there —
/// `aion update` names what it installed beside what is now running, because
/// the two are not always the same string (an `aion-cli X` can resolve an
/// `aion-server Y`, and the record carries the server's version).
///
/// # Errors
///
/// Returns [`HandoverRefusal::incomplete`] when the successor cannot be
/// started, exits during startup, or cannot be watched — every one of those
/// after the predecessor was already signalled, which the message says.
pub async fn start_successor(
    home: &Path,
    executable: &Path,
    predecessor: Option<&PidRecord>,
) -> Result<PidRecord, HandoverRefusal> {
    let log_path = home.join("server.log");
    let mut child =
        crate::launch::spawn_server_binary(executable, home, &log_path).map_err(|failure| {
            HandoverRefusal::incomplete(match predecessor {
                Some(record) => format!(
                    "server {} was signalled but the successor could not be started: {}",
                    record.pid, failure.message
                ),
                None => format!("the server could not be started: {}", failure.message),
            })
        })?;
    println!("started successor process {}", child.id());
    watch_successor(home, predecessor, &mut child, &log_path).await
}

async fn watch_successor(
    home: &Path,
    predecessor: Option<&PidRecord>,
    child: &mut std::process::Child,
    log_path: &Path,
) -> Result<PidRecord, HandoverRefusal> {
    let successor_pid = child.id();
    let mut narrator = BootNarrator::new();
    let mut predecessor_gone = predecessor.is_none();
    loop {
        match child.try_wait() {
            Ok(Some(status)) => {
                return Err(HandoverRefusal::incomplete(successor_death_report(
                    status,
                    predecessor,
                    log_path,
                )));
            }
            Ok(None) => {}
            Err(error) => {
                return Err(HandoverRefusal::incomplete(format!(
                    "could not watch the successor process: {error}; its log is {}",
                    log_path.display()
                )));
            }
        }
        if let Some(record) = predecessor
            && !predecessor_gone
            && !matches!(
                incarnation::probe(record),
                incarnation::IncarnationProbe::Verified { .. }
            )
        {
            predecessor_gone = true;
            println!(
                "server {} (version {}) has exited; the store's writer lock is free",
                record.pid, record.version
            );
        }
        if let BootObservation::Live(record) = narrator.observe(home)
            && record.pid == successor_pid
            && record.state == IncarnationState::Serving
            && let Some(http_address) = record.http_address
            && ready(http_address).await
        {
            match predecessor {
                Some(old) => println!(
                    "{} (pid {}) → {} (pid {}), serving on {http_address}",
                    old.version, old.pid, record.version, record.pid
                ),
                None => println!(
                    "{} (pid {}) is serving on {http_address}",
                    record.version, record.pid
                ),
            }
            return Ok(*record);
        }
        tokio::time::sleep(POLL_INTERVAL).await;
    }
}

/// The honest account of a successor that died during startup.
///
/// When a predecessor was drained for it, that fact leads: the old server is
/// gone, the new one is red, and this is where to look. A partial handover
/// reported as a plain start failure would leave the operator believing they
/// still have a server.
fn successor_death_report(
    status: std::process::ExitStatus,
    predecessor: Option<&PidRecord>,
    log_path: &Path,
) -> String {
    let tail = crate::launch::log_tail(log_path);
    match predecessor {
        Some(record) => format!(
            "the successor exited during startup ({status}). The previous server (pid \
             {}, version {}) was already signalled and is draining or gone, so THIS \
             HOME NOW HAS NO SERVER. Its log ends with:\n{tail}\nfull log: {}\ndeath \
             note: {}",
            record.pid,
            record.version,
            log_path.display(),
            death_note_path(log_path).display()
        ),
        None => format!(
            "the server exited during startup ({status}); its log ends with:\n{tail}\n\
             full log: {}",
            log_path.display()
        ),
    }
}

/// The death note beside a home's server log — named in a failure report so
/// the operator does not have to know the layout.
fn death_note_path(log_path: &Path) -> std::path::PathBuf {
    log_path
        .parent()
        .unwrap_or_else(|| Path::new("."))
        .join("logs")
        .join("aion-server.death.log")
}

/// Whether `GET /health/ready` answers 200 — the successor's own statement
/// that its store and runtime are up, not merely that a port accepted.
async fn ready(address: std::net::SocketAddr) -> bool {
    let Ok(client) = reqwest::Client::builder()
        .timeout(PROBE_STEP_TIMEOUT)
        .build()
    else {
        return false;
    };
    match client
        .get(format!("http://{address}/health/ready"))
        .send()
        .await
    {
        Ok(response) => response.status() == reqwest::StatusCode::OK,
        Err(_not_yet) => false,
    }
}

#[cfg(test)]
mod tests {
    use super::{HandoverRefusal, resolve_predecessor};
    use aion_server::control::{IncarnationState, PidRecord};

    type TestResult = Result<(), Box<dyn std::error::Error>>;

    fn record(state: IncarnationState) -> PidRecord {
        PidRecord {
            pid: 1234,
            started_at_unix_secs: 1,
            binary_sha256: "0".repeat(64),
            version: "0.0.0-test".to_owned(),
            commit: "test".to_owned(),
            state,
            http_address: None,
            grpc_address: None,
            intended_http_address: None,
            intended_grpc_address: None,
            stage: Some("wal-recovery".to_owned()),
            stage_detail: Some("materializing shard 3 of 8".to_owned()),
            stage_seq: 3,
            stage_updated_at_unix_secs: 1,
            drain_timeout_seconds: 30,
        }
    }

    fn plant(home: &std::path::Path, record: &PidRecord) -> TestResult {
        let run_dir = home.join("run");
        std::fs::create_dir_all(&run_dir)?;
        std::fs::write(
            run_dir.join("aion-server.pid"),
            serde_json::to_string(record)?,
        )?;
        Ok(())
    }

    /// An empty home is refused before anything is signalled, and the refusal
    /// names the verb that answers "what IS running here".
    #[test]
    fn an_empty_home_is_refused_naming_status() -> TestResult {
        let home = tempfile::tempdir()?;
        let Err(refusal) = resolve_predecessor(home.path(), "restart") else {
            return Err("an empty home must refuse a handover".into());
        };
        assert_eq!(refusal.code, 2, "a refusal before signalling is exit 2");
        assert!(
            refusal.message.contains("aion server status"),
            "the refusal must name the verb that answers: {}",
            refusal.message
        );
        Ok(())
    }

    /// A record naming a dead pid is refused: there is nothing to hand over
    /// from, and signalling a recycled pid is the whole thing these verbs
    /// exist to prevent.
    #[test]
    fn a_dead_record_is_refused() -> TestResult {
        let home = tempfile::tempdir()?;
        let mut child = std::process::Command::new("true").spawn()?;
        let dead_pid = child.id();
        child.wait()?;
        let mut dead = record(IncarnationState::Serving);
        dead.pid = dead_pid;
        dead.started_at_unix_secs = 0;
        plant(home.path(), &dead)?;
        let Err(refusal) = resolve_predecessor(home.path(), "restart") else {
            return Err("a dead record must refuse a handover".into());
        };
        assert_eq!(refusal.code, 2);
        Ok(())
    }

    /// A BOOTING incarnation is refused with the reason and the remedy: there
    /// is nothing serving to replace, and killing a boot to run the same boot
    /// again is a slower start, not a handover.
    #[test]
    fn a_booting_incarnation_is_refused_with_its_stage() -> TestResult {
        let home = tempfile::tempdir()?;
        // The only live incarnation a test can mint honestly: this process.
        let me = aion_server::control::incarnation::self_identity()?;
        let mut booting = record(IncarnationState::Booting);
        booting.pid = me.pid;
        booting.started_at_unix_secs = me.started_at_unix_secs;
        plant(home.path(), &booting)?;

        let Err(HandoverRefusal { code, message }) = resolve_predecessor(home.path(), "restart")
        else {
            return Err("a booting incarnation must refuse a handover".into());
        };
        assert_eq!(code, 2);
        assert!(message.contains("BOOTING"), "{message}");
        assert!(message.contains("wal-recovery"), "{message}");
        assert!(message.contains("aion server stop"), "{message}");
        Ok(())
    }
}