launcher/src/docker/sync.rs

use crate::config::{Cluster, Config, Node};
use crate::ssh::{run_ssh, run_ssh_long_running};
use anyhow::Result;
use tracing::info;

use super::shq;

/// Get the Docker image digest (short ID) from a node.
async fn get_image_digest(config: &Config, node: &Node, image: &str) -> Result<Option<String>> {
    let cmd = format!("docker image ls --format '{{{{.ID}}}}' {}", shq(image));
    let out = run_ssh(config, node, cmd, false).await?;
    let digest = out.trim().to_string();
    if digest.is_empty() {
        Ok(None)
    } else {
        Ok(Some(digest))
    }
}

/// Get the Docker image creation timestamp.
async fn get_image_created(config: &Config, node: &Node, image: &str) -> Option<String> {
    let cmd = format!(
        "docker image inspect --format '{{{{.Created}}}}' {} 2>/dev/null",
        shq(image)
    );
    run_ssh(config, node, cmd, false)
        .await
        .ok()
        .map(|c| c.trim().to_string())
}

/// Sync Docker image across cluster nodes.
/// Finds the node with the *most recently built* image (by creation timestamp),
/// then pipes `docker save | ssh docker load` to nodes missing or with a different digest.
///
/// **Security note:** The daisy-chain SSH pipe (`docker save | ssh -o
/// StrictHostKeyChecking=no ... docker load`) bypasses host key verification for
/// the *inner* SSH hop.  The outer SSH session (via russh) is verified, but the
/// piped `ssh` command trusts the target host unconditionally for the duration
/// of the pipe.  This is a design trade-off: re-using the verified russh session
/// for the pipe would require piping the data through the agent, which adds
/// complexity.  Acceptable for trusted-cluster deployments where the config
/// file is the trust anchor.
pub async fn sync_image_to_nodes(
    config: &Config,
    cluster: &Cluster,
    image: &str,
    explicit_source: Option<&str>,
) -> Result<()> {
    // Collect digests and creation timestamps per node.
    let mut digests: Vec<(&Node, Option<String>)> = Vec::new();
    let mut created: Vec<Option<String>> = Vec::new();
    for node in &cluster.nodes {
        let d = get_image_digest(config, node, image).await?;
        let c = if d.is_some() {
            get_image_created(config, node, image).await
        } else {
            None
        };
        digests.push((node, d));
        created.push(c);
    }

    let source_idx = if let Some(src_name) = explicit_source {
        digests
            .iter()
            .position(|(n, _)| n.name == src_name)
            .ok_or_else(|| {
                anyhow::anyhow!("sync-image-source '{}' not found in cluster", src_name)
            })?
    } else {
        let mut indexed: Vec<usize> = digests
            .iter()
            .enumerate()
            .filter(|(_, (_, d))| d.is_some())
            .map(|(i, _)| i)
            .collect();
        indexed.sort_by(|&a, &b| {
            created[b]
                .as_deref()
                .unwrap_or("")
                .cmp(created[a].as_deref().unwrap_or(""))
        });
        *indexed
            .first()
            .ok_or_else(|| anyhow::anyhow!("Image '{}' not found on any node", image))?
    };

    let (src_node, src_digest_opt) = &digests[source_idx];
    let source_digest = src_digest_opt.as_ref().unwrap();
    info!(
        "Syncing image '{}' from '{}' (most recent build) to peers...",
        image, src_node.name
    );
    for (target, target_digest) in &digests {
        if target.name == src_node.name {
            continue;
        }
        if target_digest.as_deref() == Some(source_digest) {
            info!(
                "{}",
                fluent_i18n::t!("sync-digest-match", { "name" => &target.name })
            );
            continue;
        }
        info!(
            "{}",
            fluent_i18n::t!("sync-syncing", { "name" => &target.name, "ip" => &target.ssh_ip })
        );
        let pipe_cmd = format!(
            "docker save {} | ssh -o StrictHostKeyChecking=no -i {} {}@{} docker load",
            shq(image),
            shq(&config.ssh_key_path.display().to_string()),
            shq(&config.ssh_user),
            shq(&target.ssh_ip)
        );
        run_ssh_long_running(config, src_node, pipe_cmd, false).await?;
        let verify = get_image_digest(config, target, image).await?;
        if verify.as_deref() != Some(source_digest) {
            anyhow::bail!(
                "Image sync to '{}' failed: digest mismatch (source={}, target={:?})",
                target.name,
                source_digest,
                verify
            );
        }
        info!(
            "{}",
            fluent_i18n::t!("sync-synced", { "name" => &target.name })
        );
    }
    Ok(())
}