huskies: merge 1230 story Sleds self-upgrade on startup to the gateway's published artifact if the baked binary is behind
This commit is contained in:
Huskies Agent
@@ -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::<io::watcher::WatcherEvent>(1024);
|
||||
let agents = Arc::new(AgentPool::new(port, watcher_tx.clone()));
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<String>) {
|
||||
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<String, String> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user