launcher/src/docker/launch_node.rs

//! Per-node launch body for [`start_cluster`]: resolves path references and
//! NCCL GID per node, builds the composed model dir if configured, then
//! assembles + runs the `docker run` command (or prints it in dry-run).
//!
//! Kept as a submodule so `src/docker/mod.rs` stays under the ≤500-line gate;
//! the per-node work needs SSH (per-node HF snapshots and GID indexes can
//! differ across nodes) so it must run inside the cluster spawn loop, not
//! per-cluster.

use crate::config::Node;
use crate::docker::{command, compose};

/// Per-node inputs for [`launch_node`], grouped so the function stays under
/// the argument-count lint without scattering fields.
pub(crate) struct LaunchNodeCtx<'a> {
    pub node: &'a Node,
    pub rn: &'a str,
    pub dry: bool,
    pub engine: &'a str,
    pub is_head: bool,
    pub cleanup: String,
    pub dm_ctx: &'a command::DockerCommandCtx,
    pub hf_cache: &'a std::path::Path,
    pub path_refs: &'a [command::PathRef],
}

/// Run one node's launch: returns `(node.name, rank, engine, command result)`.
pub(crate) async fn launch_node(
    config: &crate::config::Config,
    ctx: LaunchNodeCtx<'_>,
) -> (String, u32, String, anyhow::Result<String>) {
    let LaunchNodeCtx {
        node,
        rn,
        dry,
        engine,
        is_head,
        cleanup,
        dm_ctx,
        hf_cache,
        path_refs,
    } = ctx;
    let fail = |e: anyhow::Error| (node.name.clone(), node.rank, engine.to_string(), Err(e));
    if !dry && let Err(e) = crate::ssh::run_ssh(config, node, cleanup, dry).await {
        // Ignore cleanup errors — a stale container that won't die must not
        // abort the launch (the `docker run --name` will surface any real
        // conflict).
        tracing::warn!("cleanup on {}: {e}", node.name);
    }
    // Resolve model path references per node from that node's HF cache (the
    // snapshot hash can differ across nodes). Dry-run skips SSH and emits a
    // descriptive placeholder path instead. The resolved paths are substituted
    // into the recipe args/env via `{{<name>}}` before building the docker
    // command — no bind-mount is created for path refs (the HF cache is
    // already mounted at /root/.cache/huggingface, so symlinks resolve
    // correctly).
    let mut resolved_refs: Vec<(String, String)> = Vec::new();
    for spec in path_refs {
        let resolved_path = if dry {
            command::path_ref_placeholder(spec)
        } else {
            match crate::docker::path_ref::resolve_path_ref_on_node(config, node, spec, hf_cache)
                .await
            {
                Ok(p) => p,
                Err(e) => return fail(e),
            }
        };
        resolved_refs.push((spec.name.clone(), resolved_path));
    }
    let mut recipe = command::substitute_path_refs(&dm_ctx.recipe, &resolved_refs);
    let mut extra_mounts = dm_ctx.extra_mounts.clone();
    // Resolve NCCL_IB_GID_INDEX per node when the cluster config does not pin
    // `ib_gid_index` (auto-detect: scan this node's sysfs IB GID table for the
    // RoCE-v2 IPv4 GID — it re-numbers across reboots, mirroring the reference
    // serving stack). Dry-run skips SSH and renders a placeholder.
    let nccl =
        match crate::detect::resolve_nccl_for_node(config, node, dm_ctx.nccl.as_ref(), dry).await {
            Ok(n) => n,
            Err(e) => return fail(e),
        };
    // Build composed model dir if configured: symlinks into the HF cache +
    // generated files, bind-mounted at `mount`. `model` is rewritten to the
    // mount path so vLLM loads from the composed dir.
    if let Some(ref cm) = recipe.composed_model {
        match compose::build_composed_dir_on_node(config, node, rn, node.rank, cm, hf_cache, dry)
            .await
        {
            Ok((hp, cp)) => {
                extra_mounts.push(command::ResolvedMount {
                    host: hp,
                    container: cp.clone(),
                    read_only: false,
                    env: None,
                });
                recipe.model = cp;
            }
            Err(e) => return fail(e),
        }
    }
    let ctx = command::DockerCommandCtx {
        recipe,
        extra_mounts,
        nccl,
        ..dm_ctx.clone()
    };
    let cmd = match command::build_docker_command(&ctx, node, is_head) {
        Ok(c) => c.join(" "),
        Err(e) => return fail(e),
    };
    let result = crate::ssh::run_ssh_with_retry(config, node, cmd, dry, 2).await;
    (node.name.clone(), node.rank, engine.to_string(), result)
}