larust-core 0.10.0

//! Parent-side orchestration for a restart handoff: spawn a replacement
//! process, hand it the listener via `lifecycle::listener`, and wait
//! (bounded) for it to report it's actually serving before this process
//! begins its own graceful shutdown. See `readiness.rs` for the child
//! side of the same protocol.

use crate::lifecycle::{listener, supervisor};
use std::io;
use std::net::TcpListener;
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::time::Duration;
use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, Command};

/// Where a deploy step writes the path of the release that should be
/// spawned on the next restart handoff - a plain text file (not a
/// symlink: Windows symlinks need elevated privilege or Developer Mode
/// enabled, which can't be assumed), relative to the app's own root, the
/// same convention `public/`/`.env` already use elsewhere in this crate.
pub const RELEASE_POINTER_PATH: &str = "storage/releases/current";

/// Resolves which binary a restart handoff should spawn as the
/// replacement. `storage/releases/current`, if present, wins - the real
/// production deploy story: new builds land at a fresh, versioned path
/// (`storage/releases/<version-or-hash>/<name>`), and a deploy step
/// updates this pointer to the new one, atomically and auditably, with a
/// trivial rollback (just point it back). Falls back to `std::env::
/// current_exe()` - re-executing this exact running binary - when no
/// pointer file exists at all, which is only really meaningful for local
/// dev/testing (re-execing a binary the current process still holds open
/// fails outright on Windows, the same constraint `xr dev` already works
/// around by killing before rebuilding - see `docs/GOTCHAS.md`; the
/// pointer file is what a real deploy is expected to always provide
/// instead of relying on this fallback).
pub fn resolve_binary_path() -> io::Result<PathBuf> {
    resolve_binary_path_from(RELEASE_POINTER_PATH)
}

/// The testable half of `resolve_binary_path` - split out so a unit test
/// can point it at a real temp file instead of needing to change this
/// whole process's current directory (unsafe to do in a suite that runs
/// tests concurrently) just to exercise the "pointer file present" case.
fn resolve_binary_path_from(pointer_path: &str) -> io::Result<PathBuf> {
    if let Ok(contents) = std::fs::read_to_string(pointer_path) {
        let path = contents.trim();
        if !path.is_empty() {
            return Ok(PathBuf::from(path));
        }
    }
    std::env::current_exe()
}

/// The exact line `readiness::announce_ready` writes to a replacement's
/// stdout once it's genuinely about to start serving. Distinct and
/// greppable; anything else the child's own stdout produces first
/// (`tracing_subscriber`'s default formatter also writes to stdout, so
/// the app's own "starting server" log line is expected to precede this)
/// is simply skipped while scanning for this exact line, never treated
/// as a failure on its own.
pub const READY_MARKER: &str = "__LARUST_HANDOFF_READY__";

/// Spawns `binary_path` as a replacement for the process currently
/// serving on `listener`, hands the listener off to it, and waits up to
/// `ready_timeout` for it to report readiness.
///
/// - `Ok(Some(child))` - the replacement is confirmed live and serving;
///   the caller should now begin its own graceful shutdown.
/// - `Ok(None)` - the replacement crashed, exited, or never reported
///   ready within the timeout; any process that *was* spawned has
///   already been killed here, so the caller can simply keep serving as
///   if the handoff attempt never happened.
/// - `Err(_)` - couldn't even attempt the handoff (e.g. spawning
///   `binary_path` itself failed) - also always safe to just keep
///   serving.
///
/// Never partially leaves a live orphan behind: every path that doesn't
/// return `Ok(Some(child))` has already killed and reaped whatever was
/// spawned.
///
/// Stdout is deliberately `Stdio::inherit()`, not piped: only stderr
/// carries the one-line readiness handshake (see `readiness::
/// announce_ready`'s own doc comment for why routing routine logging
/// through a pipe that gets dropped once ready would otherwise break
/// every log line the replacement emits afterward). Once ready, the
/// stderr reader is kept alive and draining in the background - not just
/// dropped - for the same reason, one level up: a `tracing::warn!`/
/// `error!` call later in the replacement's life would otherwise hit the
/// exact same closed-pipe problem stdout would have.
///
/// `register_with_supervisor` distinguishes the *first* hop of a handoff
/// chain (`xr dev`/`xr restart`'s own process, which is not itself a
/// member of any job/process-group, spawning generation 1 directly) from
/// every later, server-to-server hop (generation *N* spawning generation
/// *N+1* from inside its own admin-channel handling). Pass `true` only for
/// the former. On Windows, a process that is already a member of a job
/// object automatically adds any child it spawns to that same job - no
/// further Win32 calls needed - so registering generation *N+1* into a
/// *second*, freshly-created job (as this code used to do unconditionally)
/// doesn't add protection, it *replaces* the inherited membership in `xr
/// dev`'s own job with membership in a new job whose sole owner is
/// generation *N* itself. That second job's `JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE`
/// then fires the moment generation *N* exits - including its own normal,
/// *expected* exit once its handoff to *N+1* succeeds - killing the
/// brand-new replacement as a side effect of the very success that was
/// supposed to let its predecessor retire. Confirmed empirically (not
/// assumed from docs): reproduced this exact cascade directly, generation
/// by generation, before landing this fix. See `docs/GOTCHAS.md`.
pub async fn spawn_replacement_and_wait_for_ready(
    listener: &TcpListener,
    binary_path: &Path,
    ready_timeout: Duration,
    register_with_supervisor: bool,
) -> io::Result<Option<Child>> {
    let mut command = Command::new(binary_path);
    command
        .env(listener::INHERIT_LISTENER_ENV, "1")
        .stdin(Stdio::piped())
        .stdout(Stdio::inherit())
        .stderr(Stdio::piped());
    // Guarantees the OS kills this replacement if *this* process dies for
    // any reason (crash, force-kill, closed terminal) - not just the
    // graceful paths (Ctrl+C, `STOP`) already handled elsewhere. On Linux
    // this is now armed by the replacement *itself*, inside `Application::
    // serve()`, deliberately not via a `Command::pre_exec` hook here - see
    // `lifecycle::supervisor::linux::arm_pdeathsig`'s own doc comment for
    // the real bug that design caused (forcing a hazardous raw `fork()`
    // instead of `posix_spawn()` for every handoff). Nothing to do on this
    // side at all for that mechanism anymore; `register` below still
    // covers Windows' own, entirely parent-side equivalent.

    // Unix fd inheritance is fixed at `fork()` time, which happens inside
    // `Command::spawn()` below - a duplicate fd created and CLOEXEC-
    // cleared *after* that point can never retroactively become part of
    // the already-forked child's own fd table, so on Unix this has to run
    // *before* spawning. Windows is the opposite: `WSADuplicateSocketW`
    // requires the child's real PID, which only exists *after* spawn
    // returns, so it still runs after, unchanged, further down. Confirmed
    // as a real, 100%-reproducible bug, not a theoretical one: calling
    // this after `spawn()` unconditionally (as this function used to)
    // made every single Unix restart-handoff replacement reconstruct a
    // `TcpListener` from a fd number the OS had never actually granted
    // it, aborting instantly with "fatal runtime error: IO Safety
    // violation: owned file descriptor already closed" the moment
    // anything (e.g. `try_clone()`) first touched it - not a timing- or
    // load-sensitive flake, every attempt, every time. See
    // `docs/GOTCHAS.md`.
    #[cfg(unix)]
    let pre_spawn_encoded = Some(listener::prepare_for_handoff(listener, 0)?);
    #[cfg(not(unix))]
    let pre_spawn_encoded: Option<String> = None;

    let spawn_result = command.spawn();
    // Whatever `spawn()` just did, this process's own copy of the
    // duplicate fd created above (Unix only - `pre_spawn_encoded` is
    // always `None` elsewhere) has already done its one job: `fork()`
    // already gave the child (if spawning succeeded) its own fully
    // independent copy at the same number. Left alone, this process's
    // own copy would otherwise leak for the rest of its life - one fd per
    // restart attempt, including failed ones.
    if let Some(encoded) = &pre_spawn_encoded {
        listener::close_duplicated_fd(encoded);
    }
    let mut child = spawn_result?;
    if register_with_supervisor {
        supervisor::register(&child);
    }
    let child_pid = child
        .id()
        .ok_or_else(|| io::Error::other("spawned replacement has no pid"))?;

    let encoded = match pre_spawn_encoded {
        Some(encoded) => encoded,
        None => listener::prepare_for_handoff(listener, child_pid)?,
    };
    {
        let mut stdin = child.stdin.take().expect("stdin was piped above");
        let write_result = async {
            stdin.write_all(encoded.as_bytes()).await?;
            stdin.write_all(b"\n").await
        }
        .await;
        // A replacement that crashes fast enough can already have exited -
        // closing its own stdin - before this process finishes handing it
        // the listener encoding: `BrokenPipe`, not a bug in the write
        // itself. Confirmed as a real, reachable case, not theoretical:
        // this exact path is what `tests/handoff.rs`'s own "a replacement
        // that crashes immediately" test exercises. Same outcome as every
        // other "this replacement isn't viable" branch below - report it
        // as a failed attempt, not a hard error the caller has to handle;
        // killing an already-dead process is a harmless no-op.
        if let Err(source) = write_result {
            tracing::warn!(
                error = %source,
                "restart handoff replacement's stdin closed before this process could \
                 hand off the listener - it exited or crashed immediately"
            );
            let _ = child.kill().await;
            let _ = child.wait().await;
            return Ok(None);
        }
    }

    let stderr = child.stderr.take().expect("stderr was piped above");
    let mut lines = BufReader::new(stderr).lines();

    // Captured only for the failure-diagnostic `tracing::warn!` calls below
    // - bounded small since this is meant to catch a crash's last words
    // (e.g. a panic message, a bind error), not to be a general log
    // forwarder. Not seen at all if the replacement becomes ready, the
    // overwhelmingly common case.
    let mut early_output: Vec<String> = Vec::new();
    let ready_result = tokio::time::timeout(ready_timeout, async {
        while let Ok(Some(line)) = lines.next_line().await {
            if line.trim() == READY_MARKER {
                return true;
            }
            if early_output.len() < 10 {
                early_output.push(line);
            }
        }
        false
    })
    .await;

    let became_ready = match ready_result {
        Ok(true) => true,
        // The stderr pipe hit EOF (the replacement's own process exited)
        // before ever printing the marker and before the timeout even
        // elapsed - this is a crash or an early, deliberate exit, not
        // slowness. Worth telling apart from the timeout case below: a
        // fix for "too slow to boot" (a longer timeout, less CI load) does
        // nothing for "it already exited by itself."
        Ok(false) => {
            tracing::warn!(
                ?early_output,
                "restart handoff replacement's stderr closed before it announced \
                 readiness - it exited or crashed rather than merely being slow"
            );
            false
        }
        Err(_) => {
            tracing::warn!(
                ?ready_timeout,
                ?early_output,
                "restart handoff replacement did not announce readiness within the timeout"
            );
            false
        }
    };

    if became_ready {
        // Keep draining rather than dropping `lines` here - an unread
        // stderr pipe would break the replacement's own error/warn
        // logging the same way an unread stdout pipe used to (see this
        // function's own doc comment). Forwarded to this process's own
        // stderr rather than discarded outright - `lifecycle::readiness::
        // announce_ready`'s own doc comment documents the replacement's
        // stderr as reserved for exactly one line (the handshake just
        // consumed above), so nothing *should* ever reach here in the
        // ordinary case. "Should" isn't "does": a real bug (`logging::
        // init`'s own file-open-failure diagnostic, once written via a
        // bare `eprintln!`) violated that exact contract and got silently
        // swallowed right here, with zero indication anywhere that
        // anything had gone wrong - reported directly, from a real app
        // whose log file had quietly stopped updating. Forwarding instead
        // of discarding means a future violation of the same contract
        // (this framework's own code, or - less likely, since application
        // code doesn't write to this pipe at all - anything else) stays
        // visible instead of disappearing the identical way.
        tokio::spawn(forward_drained_lines(lines, |line| eprintln!("{line}")));
        Ok(Some(child))
    } else {
        let _ = child.kill().await;
        let _ = child.wait().await;
        Ok(None)
    }
}

/// Reads every line `lines` produces until EOF, handing each one to `sink`
/// instead of discarding it - the actual behavior the doc comment at this
/// function's own call site explains. A plain generic function (not
/// inlined into the `tokio::spawn` closure directly) specifically so a
/// unit test can exercise the real drain-and-forward logic against an
/// in-memory `tokio::io::duplex()` pair instead of needing a genuine
/// spawned child process just to prove a line written after the handshake
/// completes doesn't vanish.
async fn forward_drained_lines<R: AsyncBufRead + Unpin>(
    mut lines: tokio::io::Lines<R>,
    mut sink: impl FnMut(&str),
) {
    while let Ok(Some(line)) = lines.next_line().await {
        sink(&line);
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn resolves_to_the_pointer_files_contents_when_present_and_non_empty() {
        let dir = tempfile::tempdir().unwrap();
        let pointer = dir.path().join("current");
        std::fs::write(&pointer, "  /releases/abc123/my-app  \n").unwrap();

        let resolved = resolve_binary_path_from(pointer.to_str().unwrap()).unwrap();
        assert_eq!(resolved, PathBuf::from("/releases/abc123/my-app"));
    }

    #[test]
    fn falls_back_to_current_exe_when_the_pointer_file_is_missing() {
        let dir = tempfile::tempdir().unwrap();
        let pointer = dir.path().join("does-not-exist");

        let resolved = resolve_binary_path_from(pointer.to_str().unwrap()).unwrap();
        assert_eq!(resolved, std::env::current_exe().unwrap());
    }

    #[test]
    fn falls_back_to_current_exe_when_the_pointer_file_is_empty_or_blank() {
        let dir = tempfile::tempdir().unwrap();
        let pointer = dir.path().join("current");
        std::fs::write(&pointer, "   \n").unwrap();

        let resolved = resolve_binary_path_from(pointer.to_str().unwrap()).unwrap();
        assert_eq!(resolved, std::env::current_exe().unwrap());
    }

    /// The actual regression guard: a real bug (`logging::init`'s own
    /// file-open-failure diagnostic, once written via a bare `eprintln!`)
    /// landed exactly in this drained-but-discarded stream and simply
    /// vanished, on every restart-handoff replacement, with zero
    /// indication anywhere that anything had gone wrong. Uses an in-memory
    /// `tokio::io::duplex()` pair instead of a real spawned child
    /// specifically so this is fast and deterministic - the earlier,
    /// broken version of this code (`while let Ok(Some(_)) =
    /// lines.next_line().await {}`) would fail this test immediately, by
    /// construction, not just "usually."
    #[tokio::test]
    async fn forward_drained_lines_hands_every_line_to_the_sink_instead_of_discarding_it() {
        let (mut writer, reader) = tokio::io::duplex(1024);
        let lines = tokio::io::AsyncBufReadExt::lines(BufReader::new(reader));

        let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        let seen_in_task = seen.clone();
        let task = tokio::spawn(forward_drained_lines(lines, move |line| {
            seen_in_task.lock().unwrap().push(line.to_string());
        }));

        writer
            .write_all(b"couldn't open the log file - falling back to stdout only\n")
            .await
            .unwrap();
        writer
            .write_all(b"a second unrelated line\n")
            .await
            .unwrap();
        drop(writer); // EOF - lets `forward_drained_lines` return so `task` completes.

        task.await.unwrap();
        assert_eq!(
            *seen.lock().unwrap(),
            vec![
                "couldn't open the log file - falling back to stdout only".to_string(),
                "a second unrelated line".to_string(),
            ],
            "every line written after the handshake must reach the sink, not be discarded"
        );
    }
}