aion-server 0.30.0

//! The run loop's doors: bind both transport listeners, fill the home's pid
//! record with what the binds proved, and hold the serve-until-shutdown
//! select that decides how the run ends.
//!
//! Split from `run.rs` as one coherent unit — the stop-verb work introduced
//! it whole (`OpenedDoors`, [`bind_doors`], [`serve_until_shutdown`],
//! [`transport_exit_report`]), and it owns exactly one concern: the doors and
//! what the record says once they are open.
//!
//! The CLAIM is not here any more. It happens at birth, in `run.rs`, before
//! the store opens — see [`crate::control::claim`] for why. What is left here
//! is the FILL: the addresses this boot actually bound, the drain window it
//! actually resolved, and the move out of `Booting` into `Serving`.

use tokio::net::TcpListener;

use super::ShutdownOutcome;
use super::transports::{shutdown_signal, transport_bind, transport_result};
use crate::control::{IncarnationState, PidFileGuard, RecordUpdate};
use crate::error::ServerError;
use crate::shutdown;
use crate::state::ServerState;

/// Serve until a transport exits or the first termination signal arrives,
/// returning the drain's report. A transport exit shuts the engine down and
/// reports the nothing-was-draining shape; a signal runs the full drain and
/// then reaps both transport tasks unless the drain was forced.
///
/// Both endings flip the home's record to `Draining` FIRST — before the drain
/// runs, before a transport's exit is even reported. That ordering is
/// load-bearing: `aion server restart` sends SIGTERM and then waits for the
/// record to read `Draining` before it starts the successor, because a
/// successor that birth-claims while the record still says `Serving` is
/// REFUSED as a colliding sibling. Flipping late would make restart a race.
pub(super) async fn serve_until_shutdown(
    state: &ServerState,
    pid_file_guard: &PidFileGuard,
    shutdown_tx: &tokio::sync::watch::Sender<bool>,
    grpc: &mut tokio::task::JoinHandle<Result<(), ServerError>>,
    http: &mut tokio::task::JoinHandle<Result<(), ServerError>>,
) -> Result<shutdown::ShutdownReport, ServerError> {
    Ok(tokio::select! {
        result = &mut *grpc => {
            flip_to_draining(pid_file_guard, "the gRPC transport returned");
            transport_result("gRPC", result)?;
            state.shutdown()?;
            transport_exit_report(state)
        },
        result = &mut *http => {
            flip_to_draining(pid_file_guard, "the HTTP transport returned");
            transport_result("HTTP", result)?;
            state.shutdown()?;
            transport_exit_report(state)
        },
        result = shutdown_signal() => {
            result?;
            // FIRST, before the watch send and before the drain: the record is
            // the only thing a successor can read, and every millisecond it
            // still says `Serving` is a millisecond in which `aion server
            // restart` would refuse its own successor.
            flip_to_draining(pid_file_guard, "a termination signal was observed");
            let _receiver_count = shutdown_tx.send(true);
            let report = shutdown::drain_after_first_signal(state.clone(), async {
                let _ = shutdown_signal().await;
            }).await?;
            if !matches!(report.outcome, ShutdownOutcome::Forced) {
                transport_result("gRPC", grpc.await)?;
                transport_result("HTTP", http.await)?;
            }
            report
        },
    })
}

/// Move this incarnation's record into [`IncarnationState::Draining`].
///
/// Infallible by design, like every other record write on the shutdown path:
/// the process is going down either way, and a server that refused to drain
/// because it could not describe itself would turn a bookkeeping failure into
/// an outage. A write that cannot happen is reported at `warn` with what it
/// costs (a successor's birth claim will read `Serving` and refuse).
fn flip_to_draining(pid_file_guard: &PidFileGuard, because: &str) {
    match pid_file_guard.update_own(|record| {
        record.state = IncarnationState::Draining;
        record.stage = None;
        record.stage_detail = None;
    }) {
        Ok(RecordUpdate::Written) => {
            tracing::info!(because, "home record flipped to DRAINING");
        }
        Ok(RecordUpdate::Unclaimed | RecordUpdate::NotOurs) => {}
        Err(error) => tracing::warn!(
            %error,
            because,
            "could not flip this home's pid record to DRAINING; a successor boot \
             will read it as SERVING and refuse itself as a colliding sibling. \
             Start the successor after this process exits"
        ),
    }
}

/// The opened doors: both bound transport listeners.
///
/// The pid-file guard is NOT here any more: it is created at birth and lives
/// in the run scope, spanning the whole life of the process rather than the
/// part of it that has listeners.
pub(super) struct OpenedDoors {
    pub(super) grpc_listener: TcpListener,
    pub(super) http_listener: TcpListener,
    pub(super) bound_grpc: std::net::SocketAddr,
    pub(super) bound_http: std::net::SocketAddr,
    /// Pid of this incarnation, for the outcome record at exit.
    pub(super) identity_pid: u32,
}

/// Bind BOTH transport listeners, then fill this incarnation's already-held
/// record with what the binds proved: the addresses actually bound, the drain
/// window actually resolved, and `state = Serving`.
///
/// # Errors
///
/// Returns [`ServerError::TransportBind`] when either listener cannot bind or
/// cannot report its bound address, [`ServerError::Incarnation`] when this
/// process's own identity cannot be read for the self-verify probe, and
/// [`ServerError::PidFile`] when the record cannot be updated.
pub(super) async fn bind_doors(
    pid_file_guard: &PidFileGuard,
    grpc_address: std::net::SocketAddr,
    http_address: std::net::SocketAddr,
    drain_timeout: std::time::Duration,
) -> Result<OpenedDoors, ServerError> {
    let grpc_listener = TcpListener::bind(grpc_address)
        .await
        .map_err(|source| transport_bind("grpc", grpc_address, source))?;
    let http_listener = TcpListener::bind(http_address)
        .await
        .map_err(|source| transport_bind("http", http_address, source))?;
    let bound_grpc = grpc_listener
        .local_addr()
        .map_err(|source| transport_bind("grpc", grpc_address, source))?;
    let bound_http = http_listener
        .local_addr()
        .map_err(|source| transport_bind("http", http_address, source))?;
    let update = pid_file_guard.update_own(|record| {
        record.state = IncarnationState::Serving;
        record.http_address = Some(bound_http);
        record.grpc_address = Some(bound_grpc);
        record.drain_timeout_seconds = drain_timeout.as_secs();
        // The boot is over; a stage left behind would have `status` reporting
        // a server as mid-boot forever.
        record.stage = None;
        record.stage_detail = None;
    })?;
    let pid_record = pid_file_guard.record()?;
    // Self-verify the instrument the stop/status verbs will trust: probe the
    // record just written as those verbs would. On a box where the
    // process-table start-instant is not stable (e.g. a Linux fallback
    // reading a drifting clock), every later incarnation check fails and
    // `stop` refuses to signal this healthy server while accusing the
    // operator of pid recycling — an unfixable dead-end at stop time, but a
    // plain warning at boot time.
    match crate::control::incarnation::probe(&pid_record) {
        crate::control::IncarnationProbe::Verified { .. } => {}
        other => {
            tracing::error!(
                pid = pid_record.pid,
                probe = ?other,
                "the pid record this server just wrote does not verify against its \
                 own live process: the incarnation instrument is unstable on this \
                 host, and `aion server stop`/`status` will refuse to trust the \
                 record. The server runs; the control verbs will not address it"
            );
        }
    }
    // The claim's outcome, banner-grade: one stable line an operator (or a
    // harness) can grep for, whichever face the birth claim's own reporting
    // took. An unclaimed boot serves but cannot be addressed by the control
    // verbs; that fact must not live only in a warning above the fold.
    tracing::info!(
        pid = pid_record.pid,
        home_claimed = pid_file_guard.holds_claim(),
        record_filled = matches!(update, RecordUpdate::Written),
        "home claim resolved"
    );
    Ok(OpenedDoors {
        grpc_listener,
        http_listener,
        bound_grpc,
        bound_http,
        identity_pid: pid_record.pid,
    })
}

/// The report for a run that ended because a transport task returned rather
/// than through the signal-driven drain: nothing was drained and nothing
/// parked, and the record says exactly that.
fn transport_exit_report(state: &ServerState) -> shutdown::ShutdownReport {
    shutdown::ShutdownReport {
        outcome: ShutdownOutcome::Clean,
        drain_timeout: state.runtime_config().drain_timeout,
        delivered_drain_requests: 0,
        parked: Vec::new(),
        parked_declared: Vec::new(),
        managed_workers_stopped: Vec::new(),
        managed_workers_unstopped: Vec::new(),
    }
}