//! Post-build registry push (`[registry]` config section). //! //! Images are pushed as **version tags**, not `:local`: the version is the //! image's build checksum (recursive SHA-256 of the image dir, the same //! value stamped as the `sm121.checksum` label) — identical content always //! maps to the identical tag, so re-publishing "the same thing" cannot //! create a duplicate; changed content is a new version. Each push also //! refreshes the mutable `latest` tag. //! //! A local publication ledger (`published.json` next to the global config) //! records what was published and with which image digest: a re-push with //! an unchanged digest is skipped (with a note) unless forced. The push //! itself runs in a tmux session on the build node — a 30 GB push must not //! ride on a single SSH session — polled with the same 30 s / 90 min loop //! as the build. The launcher stores no registry credentials: the node must //! be `docker login`-ed once. use super::build::{DEFAULT_MAX_BUILD_SECS, format_elapsed}; use super::shq; use crate::config::{Config, Node}; use crate::ssh::run_ssh; use anyhow::Result; use std::collections::HashMap; use tracing::info; /// One published image: its version tag, the local image digest it was /// published from, and when. Keyed by the repo name (the image name without /// its tag). #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct PublishedEntry { pub version: String, pub image_id: String, pub published_at: String, } /// Ledger file: `/published.json`. fn ledger_path() -> Option { crate::config::default_config_dir().map(|d| d.join("published.json")) } fn read_published() -> HashMap { let Some(p) = ledger_path() else { return HashMap::new(); }; std::fs::read_to_string(p) .ok() .and_then(|s| serde_json::from_str(&s).ok()) .unwrap_or_default() } fn record_published(name: &str, entry: &PublishedEntry) { let Some(p) = ledger_path() else { return; }; let mut m = read_published(); m.insert(name.to_string(), entry.clone()); if let Some(parent) = p.parent() { let _ = std::fs::create_dir_all(parent); } let _ = std::fs::write(&p, serde_json::to_string_pretty(&m).unwrap_or_default()); } /// Map a local image to the two pushed refs: `{base}/{name}:{version}` and /// `{base}/{name}:latest`. `None` when the mapping is meaningless (empty /// base, image name, or version). pub(crate) fn registry_refs( base: &str, image_tag: &str, version: &str, ) -> Option<(String, String)> { let base = base.trim().trim_end_matches('/'); let name = image_tag.trim(); let version = version.trim(); if base.is_empty() || name.is_empty() || version.is_empty() { return None; } let name_part = name.rsplit_once(':').map(|(n, _)| n).unwrap_or(name); let repo = name_part.rsplit('/').next()?.trim(); if repo.is_empty() { return None; } Some(( format!("{base}/{repo}:{version}"), format!("{base}/{repo}:latest"), )) } /// Push a built image to the configured `[registry]` (no-op when the section /// is absent). `version` = the build checksum (first 12 hex chars); both the /// version tag and `latest` are pushed. Skips (with a note) when the ledger /// records this exact image digest under the same version, unless `force`. pub async fn push_image( config: &Config, node: &Node, image_tag: &str, version: &str, force: bool, label: &str, ) -> Result<()> { let Some(registry) = config.registry.as_ref() else { return Ok(()); }; let Some((version_ref, latest_ref)) = registry_refs(®istry.base, image_tag, version) else { anyhow::bail!( "{}", fluent_i18n::t!("push-mapping-failed", { "base" => registry.base.clone(), "image" => image_tag.to_string() }) ); }; let host = registry.base.split('/').next().unwrap_or(®istry.base); let repo_name = latest_ref .rsplit_once(":latest") .map(|(repo, _)| repo.to_string()) .unwrap_or_else(|| image_tag.to_string()); // Duplicate guard: ledger says this exact image was already published // under this version → skip unless explicitly forced. let image_id = run_ssh( config, node, format!( "docker image inspect --format '{{{{.Id}}}}' {} 2>/dev/null || echo MISSING", shq(image_tag) ), false, ) .await .unwrap_or_default() .trim() .to_string(); if image_id.is_empty() || image_id == "MISSING" { let node_name = node.name.clone(); anyhow::bail!("image '{image_tag}' not found on '{node_name}' — build it first"); } if !force && let Some(e) = read_published().get(&repo_name) && e.image_id == image_id && e.version == version.trim() { info!( "{}", fluent_i18n::t!("push-already-published", { "ref" => version_ref.clone(), "label" => label.to_string() }) ); return Ok(()); } let session_name = format!("sm121-push-{label}"); let logfile = format!("/tmp/{label}-push.log"); let marker = format!("Push done: {version_ref}"); run_ssh( config, node, format!( "tmux kill-session -t {s} 2>/dev/null || true", s = shq(&session_name) ), false, ) .await?; let tmux_cmd = format!( "tmux new-session -d -s {s} \"docker push {a} > {log} 2>&1 && docker push {b} >> {log} 2>&1 \ && echo '{m}' >> {log}\"", s = shq(&session_name), a = shq(&version_ref), b = shq(&latest_ref), m = shq(&marker), log = shq(&logfile) ); info!( "{}", fluent_i18n::t!("pushing-image", { "ref" => version_ref.clone(), "session" => session_name.clone() }) ); run_ssh(config, node, tmux_cmd, false).await?; let start = std::time::Instant::now(); loop { tokio::time::sleep(std::time::Duration::from_secs(30)).await; let elapsed = start.elapsed(); if elapsed.as_secs() >= DEFAULT_MAX_BUILD_SECS { anyhow::bail!( "{}", fluent_i18n::t!("push-timeout", { "ref" => version_ref.clone(), "node" => node.name.clone(), "elapsed" => format_elapsed(elapsed), "session" => session_name.clone(), "log" => logfile.clone() }) ); } let alive = run_ssh( config, node, format!( "tmux has-session -t {s} 2>&1 || echo DEAD", s = shq(&session_name) ), false, ) .await?; let dead = alive.trim().contains("DEAD"); if let Ok(logs) = run_ssh( config, node, format!( "tail -10 {log} 2>/dev/null || echo '(no log yet)'", log = shq(&logfile) ), false, ) .await { for line in logs.lines().filter(|l| !l.is_empty()) { info!(" [push] {}", line.trim()); } } if dead { let ok = run_ssh( config, node, format!( "grep -F -- {m} {log} >/dev/null 2>&1 && echo OK || echo NO", m = shq(&marker), log = shq(&logfile) ), false, ) .await .map(|s| s.trim() == "OK") .unwrap_or(false); if ok { let digest = run_ssh( config, node, format!( "grep -oE 'digest: sha256:[0-9a-f]+ size: [0-9]+' {log} | tail -1", log = shq(&logfile) ), false, ) .await .unwrap_or_default(); let digest_suffix = digest .trim() .lines() .last() .map(|d| format!(" — {d}")) .unwrap_or_default(); info!( "{}{}", fluent_i18n::t!("pushed-image", { "ref" => version_ref.clone(), "latest" => latest_ref.clone(), "elapsed" => format_elapsed(elapsed) }), digest_suffix ); record_published( &repo_name, &PublishedEntry { version: version.trim().to_string(), image_id, published_at: crate::stamp::utc_stamp(), }, ); } else { let exit_log = run_ssh( config, node, format!("tail -20 {log}", log = shq(&logfile)), false, ) .await .unwrap_or_default(); anyhow::bail!( "{}", fluent_i18n::t!("push-failed", { "ref" => version_ref.clone(), "node" => node.name.clone(), "elapsed" => format_elapsed(elapsed), "log" => exit_log.trim().to_string(), "host" => host.to_string() }) ); } return Ok(()); } info!( "{}", fluent_i18n::t!("push-running", { "elapsed" => format_elapsed(elapsed) }) ); } }