aion_server/control/

outcome.rs

//! The shutdown outcome record: how the drain's rich result crosses the
//! process boundary.
//!
//! The process exit code deliberately collapses the drain outcome (#207:
//! `Parked` exits success because parked state is recoverable by design), so
//! the dying server writes its full [`OutcomeRecord`] into the death note —
//! the file that already owns "what ended this process" — as one `OUTCOME`
//! entry, and `aion server stop` reads it back after the process is gone.
//! One file, one writer, no second channel; the note's line-oriented format
//! means readers of the existing entries are untouched by the new kind.
//!
//! A server that dies without writing the record (`kill -9`, a panic before
//! the drain) leaves an ARMED bracket with no `OUTCOME` and no `DISARMED`;
//! the reader reports exactly that shape and never fabricates a drain
//! summary from it.

use std::path::Path;

use serde::{Deserialize, Serialize};

use crate::error::ServerError;
use crate::shutdown::ShutdownOutcome;

/// One worker's parked in-flight work at drain timeout, by name.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ParkedWorkerRecord {
    /// Registry id of the worker whose tasks were parked.
    pub worker: String,
    /// Task queue the worker was serving, when the registry still knew it.
    pub queue: Option<String>,
    /// The parked activities as `workflow/activity#attempt` names.
    pub tasks: Vec<String>,
}

/// The durable record of one shutdown's drain outcome.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct OutcomeRecord {
    /// Pid of the server that wrote the record, tying it to an ARMED bracket.
    pub pid: u32,
    /// Which face the drain ended on.
    pub outcome: ShutdownOutcome,
    /// The drain window that governed the wait, in seconds.
    pub drain_timeout_seconds: u64,
    /// How many connected workers received the drain request.
    pub delivered_drain_requests: usize,
    /// Parked in-flight work, by worker and task name. Empty on a clean
    /// drain.
    pub parked: Vec<ParkedWorkerRecord>,
    /// Declared-body commands still executing when the drain ended, as
    /// `workflow/activity#attempt` names: server-run commands no worker holds,
    /// ended with the server and re-dispatched by the next boot's replay.
    /// Defaulted on read because records written before the declared census
    /// existed — the death note is shared with older builds — carry no field,
    /// and their honest reading is "none named", not a parse refusal.
    #[serde(default)]
    pub parked_declared_commands: Vec<String>,
    /// Managed workers proven stopped (process group empty) at shutdown.
    pub managed_workers_stopped: Vec<String>,
    /// Managed workers that could NOT be proven stopped, by name — never
    /// summed into a count that could read as calm.
    pub managed_workers_unstopped: Vec<String>,
}

impl OutcomeRecord {
    /// Build the record from what the drain observed, named where the
    /// observation was made: workers by registry id, parked tasks as
    /// `workflow/activity#attempt`.
    #[must_use]
    pub fn from_report(pid: u32, report: &crate::shutdown::ShutdownReport) -> Self {
        Self {
            pid,
            outcome: report.outcome,
            drain_timeout_seconds: report.drain_timeout.as_secs(),
            delivered_drain_requests: report.delivered_drain_requests,
            parked: report
                .parked
                .iter()
                .map(|lost| ParkedWorkerRecord {
                    worker: lost.worker_id.value().to_string(),
                    queue: lost.task_queue.clone(),
                    tasks: lost
                        .tasks
                        .iter()
                        .map(|task| {
                            format!("{}/{}#{}", task.workflow_id, task.activity_id, task.attempt)
                        })
                        .collect(),
                })
                .collect(),
            parked_declared_commands: report
                .parked_declared
                .iter()
                .map(|key| format!("{}/{}#{}", key.workflow_id, key.activity_id, key.attempt))
                .collect(),
            managed_workers_stopped: report.managed_workers_stopped.clone(),
            managed_workers_unstopped: report.managed_workers_unstopped.clone(),
        }
    }
}

/// The entry keyword the record is written under in the death note.
const OUTCOME_KIND: &str = "OUTCOME";

/// Render the record as the death-note entry body (`OUTCOME <one-line json>`).
///
/// # Errors
///
/// Returns [`ServerError::DeathNote`] when the record cannot be serialized —
/// which the caller reports and survives: failing the shutdown over its own
/// receipt would invert the receipt's purpose.
pub fn render_entry(record: &OutcomeRecord) -> Result<String, ServerError> {
    let json = serde_json::to_string(record).map_err(|serialize_error| ServerError::DeathNote {
        message: format!("cannot serialize the shutdown outcome record: {serialize_error}"),
    })?;
    Ok(format!("{OUTCOME_KIND} {json}"))
}

/// What the death note says about the server incarnation that wore `pid`.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum NoteFate {
    /// No death note exists under this home — nothing was ever armed here
    /// (or the home predates the death-note era).
    NoNote,
    /// A note exists, its entries carry pid frames, and none of them is an
    /// `ARMED` for this pid.
    NoBracketForPid,
    /// The note holds entries this build cannot attribute to ANY pid: they
    /// carry no `pid=` frame, which means a server older than the per-pid
    /// framing wrote them.
    ///
    /// Reported rather than guessed. The alternative — scanning the file for
    /// a bracket and hoping the lines between belong to it — is exactly the
    /// defect the framing replaced: a refused boot's `ARMED` line landing
    /// mid-bracket made `aion server stop` report the `kill -9` shape about a
    /// clean drain. Nothing is inferred from an unframed line; the note is on
    /// disk and readable by eye, and this face says so.
    Unattributable {
        /// How many timestamped entries carry no pid frame.
        untagged_entries: usize,
    },
    /// The pid's last bracket is ARMED with no DISARMED: the process was
    /// destroyed without the run loop seeing it (`SIGKILL` class) — or is
    /// still running. The caller knows which, from its own liveness probe.
    ArmedNotDisarmed {
        /// The outcome record, if the drain got far enough to write one
        /// before the process was destroyed.
        outcome: Option<OutcomeRecord>,
        /// An `OUTCOME` line that is PRESENT in the bracket but that this
        /// binary could not parse — a torn line, or a record written by a
        /// build whose schema this one cannot read. Distinct from `outcome:
        /// None` because "the drain never wrote a record" and "the drain
        /// wrote a record I cannot read" are different facts, and reporting
        /// the second as the first sends the operator down the wrong path.
        outcome_unreadable: Option<String>,
    },
    /// The bracket closed by ordinary control flow.
    Disarmed {
        /// The drain's recorded outcome; `None` when the exit wrote no
        /// record (an error return before serving, or a pre-record binary),
        /// reported as exactly that absence.
        outcome: Option<OutcomeRecord>,
        /// An unparseable `OUTCOME` line in the bracket — the unreadable
        /// presence, kept distinct from the honest absence exactly as on
        /// [`NoteFate::ArmedNotDisarmed`].
        outcome_unreadable: Option<String>,
        /// The DISARMED line's reason text, for the report.
        reason: String,
    },
}

/// Read the death note under `home` and classify what it records for `pid`.
///
/// Entries are SELECTED by their `pid=` frame. Nothing is inferred from an
/// entry that carries no frame — that is a
/// [`NoteFate::Unattributable`] answer, not a guess.
///
/// # Errors
///
/// Returns [`ServerError::DeathNote`] when the note exists but cannot be read.
/// A malformed line inside the note is skipped, never fatal: the note is an
/// append-only log a panicking process writes into, and one torn line must
/// not make every other entry unreadable.
pub fn read_fate(home: &Path, pid: u32) -> Result<NoteFate, ServerError> {
    let path = crate::death_note::note_path(home);
    let content = match std::fs::read_to_string(&path) {
        Ok(content) => content,
        Err(io_error) if io_error.kind() == std::io::ErrorKind::NotFound => {
            return Ok(NoteFate::NoNote);
        }
        Err(io_error) => {
            return Err(ServerError::DeathNote {
                message: format!(
                    "cannot read the death note `{}`: {io_error}",
                    path.display()
                ),
            });
        }
    };
    // SELECT by pid, never scan a bracket. One home's note is shared by every
    // server that ever ran against it, and their lives interleave routinely —
    // a refused boot arms and abandons between a live server's SIGNAL and its
    // OUTCOME. Filtering first makes the interleaving irrelevant instead of
    // rare.
    let mut untagged_entries = 0_usize;
    let mut ours: Vec<&str> = Vec::new();
    for line in content.lines() {
        // A line with no timestamp is a CONTINUATION (a PANIC entry embeds a
        // multi-line backtrace) or a torn write. Neither is an entry, and
        // neither is evidence of an old format.
        let Some(body) = timestamped_body(line) else {
            continue;
        };
        match split_pid_frame(body) {
            Some((entry_pid, entry)) => {
                if entry_pid == pid {
                    ours.push(entry);
                }
            }
            None => untagged_entries += 1,
        }
    }
    // The LAST ARMED for this pid: a note outlives pid recycling, and only the
    // most recent bracket describes the incarnation a caller is asking about.
    let Some(start) = ours.iter().rposition(|entry| entry.starts_with("ARMED ")) else {
        if untagged_entries > 0 {
            return Ok(NoteFate::Unattributable { untagged_entries });
        }
        return Ok(NoteFate::NoBracketForPid);
    };
    let mut outcome: Option<OutcomeRecord> = None;
    let mut outcome_unreadable: Option<String> = None;
    let mut disarmed_reason: Option<String> = None;
    for entry in &ours[start + 1..] {
        if let Some(json) = entry.strip_prefix("OUTCOME ") {
            // A line that does not parse — torn by a kill mid-write, or
            // written by a build whose schema this binary cannot read — is
            // carried as the UNREADABLE presence, never collapsed into "no
            // record": an operator told "the drain never got far enough to
            // write one" over a record that exists is debugging the wrong
            // incident.
            // Both facts are kept in BOTH orders: a torn line beside a
            // readable record is stated by the renderers ("a readable record
            // does not un-happen the torn line beside it"), and the reader
            // must not pre-empt that rule by erasing whichever fact arrived
            // first.
            match serde_json::from_str::<OutcomeRecord>(json) {
                Ok(record) if record.pid == pid => outcome = Some(record),
                // The frame says this pid and the record says another. With
                // per-pid framing that cannot happen by interleaving, so it is
                // a torn or hand-edited line — reported, never silently
                // dropped, because a disagreement between two spellings of one
                // fact is exactly what a reader needs told.
                Ok(other) => {
                    outcome_unreadable = Some(format!(
                        "an OUTCOME line framed for pid {pid} carries a record for pid {}",
                        other.pid
                    ));
                }
                Err(parse_error) => {
                    outcome_unreadable =
                        Some(format!("an OUTCOME line does not parse: {parse_error}"));
                }
            }
        } else if let Some(reason) = entry.strip_prefix("DISARMED ") {
            disarmed_reason = Some(reason.to_owned());
            break;
        }
    }
    Ok(match disarmed_reason {
        Some(reason) => NoteFate::Disarmed {
            outcome,
            outcome_unreadable,
            reason,
        },
        None => NoteFate::ArmedNotDisarmed {
            outcome,
            outcome_unreadable,
        },
    })
}

/// Split a framed entry body into its pid and the entry itself.
///
/// The frame is `pid=<digits> <ENTRY>`. `None` for a body without one: an
/// entry written by a server older than the per-pid framing, which this build
/// reports as unattributable rather than guessing onto a pid.
fn split_pid_frame(body: &str) -> Option<(u32, &str)> {
    let tagged = body.strip_prefix("pid=")?;
    let (digits, entry) = tagged.split_once(' ')?;
    Some((digits.parse().ok()?, entry))
}

/// Strip the leading RFC 3339 timestamp from a note line, returning the
/// entry body (which still carries its `pid=` frame). `None` for lines that
/// do not carry a timestamped entry — a PANIC entry's backtrace continuation
/// lines, or a torn write.
fn timestamped_body(line: &str) -> Option<&str> {
    let (timestamp, body) = line.split_once(' ')?;
    // Cheap shape check, not a full parse: every writer-emitted line starts
    // with an RFC 3339 instant, and a torn line must simply be skipped.
    if timestamp.len() >= 20
        && timestamp
            .chars()
            .take(4)
            .all(|character| character.is_ascii_digit())
    {
        Some(body)
    } else {
        None
    }
}

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