wsb-daemon 0.258.6441

//! wsb-daemon — the process that stays.
//!
//! wsb is two programs. `wsb` is a command line: it starts, does one thing and
//! exits. `wsb-daemon` is this one: it serves the workbench so a page outlives
//! whatever opened it, keeps the machine's GitHub mirror current and carries
//! peer continuity.
//!
//! One daemon serves every worktree on the machine. It is the same program on
//! macOS, Linux and Windows.
//!
//! # What it does not do
//!
//! Work. Commands still do theirs in their own processes. This hosts the page;
//! `wsb --no-daemon` loses exactly that.

// Compiled for its tests. Its functions are called by `build.rs`, which
// `include!`s the same file because a build script cannot depend on the crate
// it builds — so nothing in the binary calls them, and that is the arrangement
// rather than an oversight.
#[allow(dead_code)]
mod build_publish;
mod config;
mod control;
mod machine;
mod peer;
mod plugin;
mod web;

use std::path::PathBuf;
use std::sync::OnceLock;

use anyhow::{Context, Result};
use config::Config;

/// Where this machine's state directory is, for the shutdown path.
///
/// Set once, before the rendezvous file is published. A daemon that stopped
/// without removing it would leave every command line on this machine pointing
/// at a port nothing answers, until one of them noticed and cleaned up.
static STATE_DIR: OnceLock<PathBuf> = OnceLock::new();

/// Everything before the runtime, then the runtime.
///
/// The already-running check talks to whatever holds the machine's port with
/// the same blocking client a command line uses, and a blocking client is not
/// something to run inside an async runtime. It happens here, before there is
/// one.
fn main() -> Result<()> {
    // Answered before anything else starts, and before the tracing subscriber
    // writes a line: this is the one question a daemon can be asked without
    // becoming one.
    //
    // The two halves of wsb are installed together and read the same database
    // schema, so "which version is this one" is the question you ask when they
    // disagree — and `wsb upgrade`, which replaces both or neither, asks it of
    // each built binary before installing either. A binary that cannot start
    // compiles perfectly well; running it is what finds out.
    if std::env::args().skip(1).any(|argument| argument == "--version") {
        println!("wsb-daemon {}", wsb::get_version());
        return Ok(());
    }

    let config = Config::from_args(std::env::args().skip(1)).map_err(|e| anyhow::anyhow!(e))?;

    // `--debug` is what `wsb serve --debug` asks for; RUST_LOG, when it is
    // set, says exactly what to log and wins over both.
    {
        use tracing_subscriber::util::SubscriberInitExt;
        subscriber(config.debug).init();
    }

    let machine_dir = wsb::daemon::machine_dir().map_err(|e| anyhow::anyhow!(e))?;
    let _ = STATE_DIR.set(machine_dir.clone());

    // Out of whatever directory it was started in, before it serves anything.
    //
    // A daemon serves whichever workspace a command's directory resolves to,
    // and resolving one asks git to discover a repository from an absolute
    // path — which fails outright when the PROCESS's own working directory has
    // been deleted underneath it. A daemon started inside a test's temporary
    // workspace then outlived that directory and answered every command on the
    // machine with "/home/…/machinefabric is not inside a git repository", for
    // a repository that plainly is one. `/` cannot be removed.
    std::env::set_current_dir("/").context("cannot leave the starting directory")?;

    // Refuse to start beside a daemon that is already serving this machine.
    // Two would hold two identities, and a command line
    // would reach whichever one published its file last.
    refuse_if_already_running(&machine_dir)?;

    tracing::info!(
        machine_dir = %machine_dir.display(),
        listen = %config.listen,
        port = config.http_port,
        peer_listen = %config.peer_listen,
        peer_port = config.peer_port,
        version = wsb::get_version(),
        "wsb-daemon starting"
    );

    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .context("cannot start the daemon's runtime")?
        .block_on(serve(config, machine_dir))
}

/// The daemon proper: hold the machine's state, serve the workbench, listen
/// for peers, publish where to find it, and take the rendezvous file with it
/// on the way out.
async fn serve(config: Config, machine_dir: PathBuf) -> Result<()> {
    let machine = machine::MachineState::new(
        machine_dir.clone(),
        config.listen,
        config.http_port,
        config.peer_listen,
        config.peer_port,
    )?;

    // A project's shared state lives under `$PRJ/.state/projects/`; an older
    // wsb kept it in the project folder, which holds worktrees only. Moved
    // before anything reads it.
    match wsb_prj::Prj::from_env() {
        Ok(prj) => {
            for line in prj.migrate_project_state().map_err(|e| anyhow::anyhow!(e))? {
                tracing::info!("{line}");
            }
        }
        Err(wsb_prj::PrjError::NotSet) => {}
        Err(error) => return Err(anyhow::anyhow!(error)),
    }

    // Records of worktrees that are not worktrees are corrected before any
    // of them is served — see `forget_contained_worktrees`.
    for root in machine.forget_contained_worktrees().map_err(|e| anyhow::anyhow!(e))? {
        tracing::info!(
            root = %root.display(),
            "forgot a recorded worktree that is inside another one"
        );
    }

    // A worktree named at startup is opened before the page is served, so
    // the first thing a person sees is not an empty machine — with its
    // services, as every other worktree gets them when it is first resolved.
    //
    // RESOLVED, not registered: the daemon is started from whatever directory
    // the command that wanted it was run in, and that is routinely a submodule
    // of the worktree rather than its root.
    if let Some(root) = &config.workspace_root {
        match machine.resolve(root).await {
            Ok(runtime) => {
                web::spawn_worktree_services_once(&runtime);
                tracing::info!(
                    worktree = %runtime.registered.id,
                    "opened the worktree this daemon was started in"
                )
            }
            Err(error) => tracing::warn!(
                %error,
                root = %root.display(),
                "the worktree named at startup could not be opened"
            ),
        }
    }

    // The rendezvous file is published only once the port is CLAIMED. A file
    // published earlier sends command lines at a port nothing is listening on,
    // and they would each wait out their start timeout to discover it.
    let (bound, is_bound) = tokio::sync::oneshot::channel();
    let server = tokio::spawn(web::start_server(machine.clone(), config.parent_pid, bound));

    let http_bound = match is_bound.await {
        Ok(address) => address,
        Err(_) => {
            let failure = server
                .await
                .context("the workbench server ended before it started")?;
            failure.context("the workbench server could not start")?;
            anyhow::bail!("the workbench server ended before it started, without an error");
        }
    };

    // Peers are served on their own threads: the protocol is blocking, and so
    // is the work behind it — scanning trees, hashing files, writing them.
    // Running that on the runtime's threads would starve the workbench.
    let peer_listener = if config.peer_port == 0 {
        tracing::info!("peer listening is off (--peer-port 0)");
        None
    } else {
        match machine.peers.listen() {
            Ok(listener) => Some(listener),
            Err(error) => {
                // Not fatal: a machine whose peer port is taken still serves
                // its workbench and still runs commands. It says so loudly,
                // because continuity will not work until it is fixed.
                tracing::error!(%error, "peers cannot reach this machine");
                None
            }
        }
    };
    // What peers can actually reach, which is not what was asked for.
    //
    // Zero when there is no listener, and that is the same zero `--peer-port 0`
    // means: nothing is listening. Publishing the REQUESTED port when the bind
    // failed told every reader — a person running `peer status`, the other
    // machine's operator reading an address off a screen — that peers could
    // reach a port nothing was on.
    let peer_bound = match &peer_listener {
        Some(listener) => listener
            .local_addr()
            .context("the peer listener has no address")?
            .port(),
        None => 0,
    };
    if let Some(listener) = peer_listener {
        peer::service::spawn_accept_loop(machine.clone(), listener);
        peer::coordinator::spawn_reconcile_loop(machine.clone(), tokio::runtime::Handle::current());
    }

    wsb::daemon::MachineRendezvous {
        version: wsb::get_version().to_string(),
        pid: std::process::id(),
        http_address: config.listen,
        http_port: http_bound.port(),
        machine_id: machine.identity.id().to_string(),
        machine_name: machine.identity.name().to_string(),
        peer_address: config.peer_listen,
        peer_port: peer_bound,
        started_at: machine.started_at,
        // Minted here, once, and never written anywhere but the 0600
        // rendezvous file. It is what lets the operator's own command line
        // stop this daemon.
        control_token: machine.control_token.clone(),
    }
    .write(&machine_dir)
    .map_err(|e| anyhow::anyhow!(e))?;

    // Ordinary termination — a Ctrl-C, a system shutdown — takes the
    // rendezvous file with it. `shutdown` in the control API does the same for
    // the path that does not come back through here.
    let stopping = tokio::spawn(async {
        if tokio::signal::ctrl_c().await.is_ok() {
            remove_rendezvous();
            tracing::info!("interrupted; stopping");
            std::process::exit(0);
        }
    });

    let outcome = server.await.context("the workbench server ended")?;
    stopping.abort();
    remove_rendezvous();
    outcome.context("the workbench server failed")
}

/// What the daemon logs through: its console, and the process's log stream,
/// which the workbench's Output panel reads.
///
/// `--debug` is what `wsb serve --debug` asks for; RUST_LOG, when it is set,
/// says exactly what to log and wins over both. The filter is the console's:
/// the stream applies its own, so a quiet console still feeds the panel the
/// daemon's information.
fn subscriber(debug: bool) -> impl tracing::Subscriber + Send + Sync {
    use tracing_subscriber::layer::SubscriberExt;
    use tracing_subscriber::Layer;
    let default_level = if debug { "debug" } else { "info" };
    let console = tracing_subscriber::fmt::layer().with_filter(
        tracing_subscriber::EnvFilter::try_from_default_env()
            // refinery prints every migration it applies, whole: thousands
            // of lines of SQL in front of whatever the daemon had to say.
            .unwrap_or_else(|_| {
                tracing_subscriber::EnvFilter::new(format!("{default_level},refinery=warn,refinery_core=warn"))
            }),
    );
    tracing_subscriber::registry()
        .with(console)
        .with(wsb::logging::BroadcastLayer.with_filter(if debug {
            tracing_subscriber::filter::LevelFilter::DEBUG
        } else {
            tracing_subscriber::filter::LevelFilter::INFO
        }))
}

/// Remove this daemon's rendezvous file — if it is still this daemon's.
///
/// A daemon that lost a race to bind, or was started beside a live one, exits
/// through here too, and the file then names the daemon that is serving.
///
/// Best effort by necessity: this runs on the way out, and a state directory
/// deleted underneath a running daemon is not something a stopping process can
/// do anything about. A file left behind is recovered from — the next command
/// finds its process gone and replaces it.
pub(crate) fn remove_rendezvous() {
    if let Some(machine_dir) = STATE_DIR.get() {
        if let Err(error) = wsb::daemon::MachineRendezvous::remove_if_published_by(machine_dir, std::process::id()) {
            tracing::warn!(%error, "could not remove the rendezvous file");
        }
    }
}

/// Stop if this machine already has a daemon that answers.
fn refuse_if_already_running(machine_dir: &std::path::Path) -> Result<()> {
    let Some(existing) =
        wsb::daemon::MachineRendezvous::read(machine_dir).map_err(|e| anyhow::anyhow!(e))?
    else {
        // Nothing this build can read — which is not the same as nothing
        // running. A daemon of another build holds the port and answers an
        // API this one cannot parse, so it is found by its pid instead.
        // Binding "the first free port" beside it would put two daemons on one
        // machine, which is the state this check exists to prevent.
        if let Some(published) =
            wsb::daemon::MachineRendezvous::published_by(machine_dir).map_err(|e| anyhow::anyhow!(e))?
        {
            if wsb::ops::process::is_alive(published.pid) {
                anyhow::bail!(
                    "wsb-daemon {} is already serving this machine (pid {}), and this build \
                     cannot read the file it published.\n                       One daemon serves one machine. Stop it first:\n                           wsb daemon stop",
                    published.version,
                    published.pid
                );
            }
            // Its process is gone, so its file is stale whatever build wrote
            // it: replaced rather than reported.
            wsb::daemon::MachineRendezvous::remove_if_published_by(machine_dir, published.pid)
                .map_err(|e| anyhow::anyhow!(e))?;
        }
        return Ok(());
    };
    let client = wsb::daemon::Client::from_rendezvous(&existing)
        .map_err(|e| anyhow::anyhow!(e))?;
    match client.info() {
        Ok(info) => anyhow::bail!(
            "wsb-daemon {} is already serving this machine on port {} (pid {}).\n  \
             One daemon serves one machine. Stop it first:\n      \
             wsb daemon stop",
            info.version,
            info.http_port,
            info.pid
        ),
        // No answer is not the same as nobody there. A daemon too busy to
        // answer still holds its port and its process; starting beside it put
        // two daemons on one machine. Stale only when its process is gone or
        // its port is free — then removed rather than reported, because a
        // machine that has been rebooted is not a machine with a problem, and
        // this daemon takes its place.
        Err(_) => {
            if wsb::daemon::launch::still_serving(&existing) {
                anyhow::bail!(
                    "wsb-daemon (pid {}) is serving this machine on port {} but did not answer: it \
                     is busy, not gone.\n  One daemon serves one machine. Wait for it, or stop it \
                     first:\n      wsb daemon stop",
                    existing.pid,
                    existing.http_port
                );
            }
            wsb::daemon::MachineRendezvous::remove_if_published_by(machine_dir, existing.pid)
                .map_err(|e| anyhow::anyhow!(e))?;
            Ok(())
        }
    }
}

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