diff --git a/server/src/main.rs b/server/src/main.rs index 4a2edb73..3dc45a3b 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -199,6 +199,15 @@ async fn main() -> Result<(), std::io::Error> { return agent_mode::run(agent_root, rendezvous, port, join_token, agent_gateway_url).await; } + // Sled startup self-upgrade (story 1230): if the gateway has published a + // newer artifact than this baked binary, swap to it and restart before + // this process ever serves a request. Never blocks startup on failure. + let self_upgrade_root = app_state.project_root.lock().unwrap().clone(); + if let Some(root) = self_upgrade_root { + let self_upgrade_gateway_url = std::env::var("HUSKIES_GATEWAY_URL").ok(); + startup::self_upgrade::maybe_self_upgrade(&root, self_upgrade_gateway_url).await; + } + // Event bus: broadcast channel for pipeline lifecycle events. let (watcher_tx, _) = broadcast::channel::(1024); let agents = Arc::new(AgentPool::new(port, watcher_tx.clone())); diff --git a/server/src/startup/mod.rs b/server/src/startup/mod.rs index ed7a4534..26ae0171 100644 --- a/server/src/startup/mod.rs +++ b/server/src/startup/mod.rs @@ -2,4 +2,7 @@ pub(crate) mod bots; pub(crate) mod project; +/// Sled startup self-upgrade — check the gateway's published artifact hash +/// and swap in place before serving if the baked binary is behind. +pub(crate) mod self_upgrade; pub(crate) mod tick_loop; diff --git a/server/src/startup/self_upgrade.rs b/server/src/startup/self_upgrade.rs new file mode 100644 index 00000000..f41ec2c8 --- /dev/null +++ b/server/src/startup/self_upgrade.rs @@ -0,0 +1,339 @@ +//! Sled startup self-upgrade — check the gateway's published artifact hash +//! against this binary's own build hash and swap in place before serving. + +use crate::slog; +use std::path::Path; +use std::time::Duration; + +/// Name of the on-disk marker recording the target hash of the last upgrade +/// attempt, used to avoid retrying forever if the artifact never converges. +const ATTEMPT_MARKER: &str = ".huskies_self_upgrade_attempt"; + +/// What startup should do after comparing the gateway's published artifact +/// hash against this binary's own build hash. +#[derive(Debug, PartialEq)] +enum SelfUpgradeDecision { + /// Gateway not configured, unreachable, or hash missing/invalid. + Skip(String), + /// Already on the published build. + UpToDate, + /// Hash differs and no prior attempt targeted this exact hash — upgrade. + Upgrade { expected_hash: String }, + /// Hash still differs after a prior attempt already targeted this exact + /// hash — stop instead of looping. + AlreadyAttempted { expected_hash: String }, +} + +/// Check the gateway's published artifact hash and self-upgrade in place if +/// the baked binary is behind. +/// +/// Called once from `main()` for standard "sled" mode, before the HTTP +/// server starts serving. `gateway_url` is the caller's already-resolved +/// `HUSKIES_GATEWAY_URL` (plain HTTP base, e.g. +/// `http://host.docker.internal:3000`) — taken as a parameter rather than +/// read from the env directly so tests can drive this deterministically +/// without mutating shared process-global env state (env vars aren't +/// per-test-isolated; this function is exercised concurrently with the rest +/// of the suite). Any failure to reach the gateway or resolve a valid hash +/// is logged as a warning and startup continues on the baked binary — this +/// check must never hard-fail startup. On a genuine mismatch it fetches and +/// swaps the binary, then exits so Docker restarts the container into the +/// new build (mirrors `/api/upgrade`, `server/src/upgrade.rs`). +pub async fn maybe_self_upgrade(project_root: &Path, gateway_url: Option) { + let Some(gateway_url) = gateway_url else { + slog!("[self-upgrade] HUSKIES_GATEWAY_URL not set; skipping startup self-upgrade check."); + return; + }; + + let current_hash = option_env!("BUILD_GIT_HASH").unwrap_or("unknown"); + let base = gateway_url.trim_end_matches('/'); + let hash_url = format!( + "{base}/api/artifacts/{}.hash", + crate::http::SLED_ARTIFACT_NAME + ); + let marker_path = project_root.join(ATTEMPT_MARKER); + + let client = match reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build() + { + Ok(c) => c, + Err(e) => { + slog!("[self-upgrade] Failed to build HTTP client: {e}; booting on baked binary."); + return; + } + }; + + match resolve_decision(&client, &hash_url, current_hash, &marker_path).await { + SelfUpgradeDecision::Skip(reason) => { + slog!("[self-upgrade] {reason}; booting on baked binary ({current_hash})."); + } + SelfUpgradeDecision::UpToDate => { + let _ = std::fs::remove_file(&marker_path); + slog!("[self-upgrade] Already on published build ({current_hash}); no upgrade needed."); + } + SelfUpgradeDecision::AlreadyAttempted { expected_hash } => { + slog!( + "[self-upgrade] Already attempted upgrade to {expected_hash} and it did not \ + converge (still on {current_hash}); not retrying. Booting on baked binary." + ); + } + SelfUpgradeDecision::Upgrade { expected_hash } => { + slog!( + "[self-upgrade] Baked binary ({current_hash}) is behind published artifact \ + ({expected_hash}); upgrading before serving." + ); + let _ = std::fs::write(&marker_path, &expected_hash); + + let artifact_url = format!("{base}/api/artifacts/{}", crate::http::SLED_ARTIFACT_NAME); + let target = crate::upgrade::resolve_target_path(); + if let Err(e) = crate::upgrade::fetch_and_replace_binary(&artifact_url, &target).await { + slog!( + "[self-upgrade] Failed to fetch/replace binary: {e}; booting on baked binary." + ); + return; + } + + crate::rebuild::drain_and_exit(project_root, "self-upgrade").await; + } + } +} + +/// Decide what to do given the current state, without any process-exiting +/// side effects — kept separate from [`maybe_self_upgrade`] so the decision +/// logic is unit-testable. +async fn resolve_decision( + client: &reqwest::Client, + hash_url: &str, + current_hash: &str, + marker_path: &Path, +) -> SelfUpgradeDecision { + let expected_hash = match fetch_expected_hash(client, hash_url).await { + Ok(h) => h, + Err(e) => return SelfUpgradeDecision::Skip(e), + }; + + if expected_hash == current_hash { + return SelfUpgradeDecision::UpToDate; + } + + if let Ok(prev) = std::fs::read_to_string(marker_path) + && prev.trim() == expected_hash + { + return SelfUpgradeDecision::AlreadyAttempted { expected_hash }; + } + + SelfUpgradeDecision::Upgrade { expected_hash } +} + +/// Fetch and validate the `.hash` sidecar from `hash_url`. +/// +/// Returns `Err` with a human-readable reason for any of: connect failure, +/// non-2xx response, empty body, or content that doesn't look like a git +/// short hash (hex digits only). +async fn fetch_expected_hash(client: &reqwest::Client, hash_url: &str) -> Result { + let resp = client + .get(hash_url) + .send() + .await + .map_err(|e| format!("gateway unreachable at {hash_url}: {e}"))?; + + if !resp.status().is_success() { + return Err(format!( + "hash fetch returned HTTP {} from {hash_url}", + resp.status() + )); + } + + let text = resp + .text() + .await + .map_err(|e| format!("failed to read hash response from {hash_url}: {e}"))?; + let hash = text.trim().to_string(); + + if hash.is_empty() { + return Err(format!("hash sidecar at {hash_url} was empty")); + } + if !hash.chars().all(|c| c.is_ascii_hexdigit()) { + return Err(format!( + "hash sidecar at {hash_url} did not look like a git hash: {hash:?}" + )); + } + + Ok(hash) +} + +// ── Tests ──────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + + /// Start a tiny HTTP server that serves `body` with `status` at `/`. + async fn serve_text(status: u16, body: &'static str) -> (u16, tokio::task::JoinHandle<()>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + + let handle = tokio::spawn(async move { + loop { + let Ok((mut stream, _)) = listener.accept().await else { + break; + }; + tokio::spawn(async move { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let mut buf = [0u8; 4096]; + let _ = stream.read(&mut buf).await; + let status_line = match status { + 200 => "200 OK", + 404 => "404 Not Found", + _ => "500 Internal Server Error", + }; + let header = format!( + "HTTP/1.1 {status_line}\r\nContent-Length: {}\r\n\r\n", + body.len() + ); + let _ = stream.write_all(header.as_bytes()).await; + let _ = stream.write_all(body.as_bytes()).await; + }); + } + }); + + (port, handle) + } + + fn test_client() -> reqwest::Client { + reqwest::Client::builder() + .timeout(Duration::from_secs(5)) + .build() + .unwrap() + } + + // ── resolve_decision ───────────────────────────────────────────────── + + #[tokio::test] + async fn up_to_date_when_hash_matches() { + let (port, _srv) = serve_text(200, "abc1234").await; + let url = format!("http://127.0.0.1:{port}/hash"); + let dir = tempfile::tempdir().unwrap(); + let marker = dir.path().join(ATTEMPT_MARKER); + + let decision = resolve_decision(&test_client(), &url, "abc1234", &marker).await; + assert_eq!(decision, SelfUpgradeDecision::UpToDate); + } + + #[tokio::test] + async fn upgrade_when_hash_differs_and_no_prior_attempt() { + let (port, _srv) = serve_text(200, "def5678").await; + let url = format!("http://127.0.0.1:{port}/hash"); + let dir = tempfile::tempdir().unwrap(); + let marker = dir.path().join(ATTEMPT_MARKER); + + let decision = resolve_decision(&test_client(), &url, "abc1234", &marker).await; + assert_eq!( + decision, + SelfUpgradeDecision::Upgrade { + expected_hash: "def5678".to_string() + } + ); + } + + #[tokio::test] + async fn already_attempted_when_marker_matches_expected_hash() { + let (port, _srv) = serve_text(200, "def5678").await; + let url = format!("http://127.0.0.1:{port}/hash"); + let dir = tempfile::tempdir().unwrap(); + let marker = dir.path().join(ATTEMPT_MARKER); + std::fs::write(&marker, "def5678").unwrap(); + + let decision = resolve_decision(&test_client(), &url, "abc1234", &marker).await; + assert_eq!( + decision, + SelfUpgradeDecision::AlreadyAttempted { + expected_hash: "def5678".to_string() + } + ); + } + + #[tokio::test] + async fn upgrade_retried_when_marker_names_a_different_hash() { + // A stale marker from a previous, different target hash must not + // block upgrading to a newly-published one. + let (port, _srv) = serve_text(200, "cafef00d").await; + let url = format!("http://127.0.0.1:{port}/hash"); + let dir = tempfile::tempdir().unwrap(); + let marker = dir.path().join(ATTEMPT_MARKER); + std::fs::write(&marker, "deadbeef").unwrap(); + + let decision = resolve_decision(&test_client(), &url, "abc1234", &marker).await; + assert_eq!( + decision, + SelfUpgradeDecision::Upgrade { + expected_hash: "cafef00d".to_string() + } + ); + } + + #[tokio::test] + async fn skip_when_gateway_unreachable() { + let dir = tempfile::tempdir().unwrap(); + let marker = dir.path().join(ATTEMPT_MARKER); + let decision = resolve_decision( + &test_client(), + "http://127.0.0.1:1/hash", + "abc1234", + &marker, + ) + .await; + assert!(matches!(decision, SelfUpgradeDecision::Skip(_))); + } + + #[tokio::test] + async fn skip_when_hash_endpoint_404s() { + let (port, _srv) = serve_text(404, "not found").await; + let url = format!("http://127.0.0.1:{port}/hash"); + let dir = tempfile::tempdir().unwrap(); + let marker = dir.path().join(ATTEMPT_MARKER); + + let decision = resolve_decision(&test_client(), &url, "abc1234", &marker).await; + assert!(matches!(decision, SelfUpgradeDecision::Skip(_))); + } + + #[tokio::test] + async fn skip_when_hash_body_is_empty() { + let (port, _srv) = serve_text(200, "").await; + let url = format!("http://127.0.0.1:{port}/hash"); + let dir = tempfile::tempdir().unwrap(); + let marker = dir.path().join(ATTEMPT_MARKER); + + let decision = resolve_decision(&test_client(), &url, "abc1234", &marker).await; + assert!(matches!(decision, SelfUpgradeDecision::Skip(_))); + } + + #[tokio::test] + async fn skip_when_hash_body_is_not_hex() { + let (port, _srv) = serve_text(200, "not-a-hash!!").await; + let url = format!("http://127.0.0.1:{port}/hash"); + let dir = tempfile::tempdir().unwrap(); + let marker = dir.path().join(ATTEMPT_MARKER); + + let decision = resolve_decision(&test_client(), &url, "abc1234", &marker).await; + assert!(matches!(decision, SelfUpgradeDecision::Skip(_))); + } + + // ── maybe_self_upgrade (non-exiting paths only) ───────────────────── + + #[tokio::test] + async fn maybe_self_upgrade_skips_when_gateway_url_none() { + let dir = tempfile::tempdir().unwrap(); + // Must return (not hang, not exit) when the gateway isn't configured. + maybe_self_upgrade(dir.path(), None).await; + } + + #[tokio::test] + async fn maybe_self_upgrade_no_ops_when_already_up_to_date() { + let current_hash = option_env!("BUILD_GIT_HASH").unwrap_or("unknown"); + let (port, _srv) = serve_text(200, current_hash).await; + let dir = tempfile::tempdir().unwrap(); + maybe_self_upgrade(dir.path(), Some(format!("http://127.0.0.1:{port}"))).await; + } +}