crashlabs/huskies
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 37 Wiki Activity

Add upgrade all chat command with drain check and convergence verify

Browse Source 
- `upgrade all` sweeps every registered sled in sequence, streaming
  per-sled phase markers and reporting a summary.
- Binary source is now the gateway's own artifact store via
  host.docker.internal (was: unresolvable `gateway` hostname serving
  the gateway's macOS binary to Linux sleds — would have bricked them).
- Sleds with active claude processes are skipped, never killed.
- After reconnect, /api/version git_hash is compared against the
  published artifact's .hash sidecar; divergence is reported loudly.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019fHdm92yjvguPi2LiXfLB9
This commit is contained in:
Timmy
2026-07-15 16:22:11 +01:00
co-authored by Claude Fable 5
parent 07b9e1605d
commit 18ba57a7b0
4 changed files with 250 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/chat/transport/matrix/bot/messages/on_room_message.rs
+49 -2
View File
@@ -537,7 +537,7 @@ pub(in crate::chat::transport::matrix::bot) async fn on_room_message(
let bot_sent = Arc::clone(&ctx.bot_sent_event_ids);
let room = room_id_str.clone();
let response = super::super::super::sled_upgrade::handle_sled_upgrade(
let outcome = super::super::super::sled_upgrade::handle_sled_upgrade(
&project,
store,
ctx.gateway_port,
@@ -557,6 +557,53 @@ pub(in crate::chat::transport::matrix::bot) async fn on_room_message(
},
)
.await;
let response = format!("{project}: {outcome}");
let html = markdown_to_html(&response);
if let Ok(msg_id) = ctx
.transport
.send_message(&room_id_str, &response, &html)
.await
&& let Ok(event_id) = msg_id.parse()
{
ctx.bot_sent_event_ids.lock().await.insert(event_id);
}
} else {
let msg = "Gateway projects store unavailable — cannot upgrade sled.";
let html = markdown_to_html(msg);
if let Ok(msg_id) = ctx.transport.send_message(&room_id_str, msg, &html).await
&& let Ok(event_id) = msg_id.parse()
{
ctx.bot_sent_event_ids.lock().await.insert(event_id);
}
}
}
super::super::super::sled_upgrade::UpgradeCommand::UpgradeAll => {
slog!("[matrix-bot] Handling 'upgrade all' from {sender}");
if let Some(ref store) = ctx.gateway_projects_store {
let transport = Arc::clone(&ctx.transport);
let bot_sent = Arc::clone(&ctx.bot_sent_event_ids);
let room = room_id_str.clone();
let response = super::super::super::sled_upgrade::handle_upgrade_all(
store,
ctx.gateway_port,
|phase_msg| {
let transport = Arc::clone(&transport);
let bot_sent = Arc::clone(&bot_sent);
let room = room.clone();
async move {
let html = markdown_to_html(&phase_msg);
if let Ok(msg_id) =
transport.send_message(&room, &phase_msg, &html).await
&& let Ok(event_id) = msg_id.parse()
{
bot_sent.lock().await.insert(event_id);
}
}
},
)
.await;
let html = markdown_to_html(&response);
if let Ok(msg_id) = ctx
@@ -568,7 +615,7 @@ pub(in crate::chat::transport::matrix::bot) async fn on_room_message(
ctx.bot_sent_event_ids.lock().await.insert(event_id);
}
} else {
let msg = "Gateway projects store unavailable — cannot upgrade sled.";
let msg = "Gateway projects store unavailable — cannot upgrade sleds.";
let html = markdown_to_html(msg);
if let Ok(msg_id) = ctx.transport.send_message(&room_id_str, msg, &html).await
&& let Ok(event_id) = msg_id.parse()
server/src/chat/transport/matrix/project_rebuild.rs
+1 -1
View File
@@ -343,7 +343,7 @@ async fn wait_for_drain(container_name: &str, timeout_secs: u64) -> Option<Strin
///
/// Uses `docker exec <name> pgrep -f claude` — exits 0 with PID list when found,
/// exits 1 when no matches (treated as 0 active processes).
async fn count_active_claude_processes(container_name: &str) -> Result<usize, String> {
pub(crate) async fn count_active_claude_processes(container_name: &str) -> Result<usize, String> {
let out = tokio::process::Command::new("docker")
.args(["exec", container_name, "pgrep", "-f", "claude"])
.output()
server/src/chat/transport/matrix/sled_upgrade.rs
+193 -67
View File
@@ -1,16 +1,26 @@
//! `upgrade [<project>]` gateway chat command — streaming sled binary upgrade.
//! `upgrade [<project>|all]` gateway chat command — streaming sled binary upgrade.
//!
//! Usage (gateway mode only):
//! - `{bot} upgrade <project>` — upgrade the named sled's binary in-container.
//! - `{bot} upgrade all` — upgrade every registered sled in sequence.
//! - `{bot} upgrade` — list registered projects (shows what can be targeted).
//!
//! The gateway orchestrates the upgrade in four phases, streaming a marker to
//! The binary comes from the gateway's own artifact store
//! (`~/.huskies/artifacts/`, published by the `release` command) — sleds never
//! download from anywhere but their gateway. Sleds with active agent
//! processes are skipped so an upgrade never kills in-flight work.
//!
//! The gateway orchestrates each upgrade in four phases, streaming a marker to
//! the chat room at each step:
//! 1. `[1/4] downloading` — POSTs to `{sled_url}/api/upgrade`; sled starts download.
//! 2. `[2/4] swapping binary` — gateway received 202; sled atomically renamed the binary.
//! 3. `[3/4] restarting sled` — sled re-execs with the new binary; HTTP goes dark briefly.
//! 3. `[3/4] restarting sled` — sled exits cleanly; Docker restarts it with the new binary.
//! 4. `[4/4] reconnected to gateway` — sled's `/health` probe is responding again.
//!
//! After reconnection the gateway polls `/api/version` and verifies the sled's
//! reported git hash matches the published artifact's hash (when a `.hash`
//! sidecar file exists).
//!
//! Concurrent `upgrade` invocations are serialised via a global async mutex so
//! that two simultaneous upgrades cannot interleave their phase markers or race
//! on the sled restart.
@@ -40,6 +50,8 @@ pub enum UpgradeCommand {
/// The project/sled name to upgrade.
project: String,
},
/// `upgrade all` — upgrade every registered sled in sequence.
UpgradeAll,
/// `upgrade` with no argument — list available projects.
ListProjects,
}
@@ -68,10 +80,14 @@ pub fn extract_upgrade_command(
}
if rest.is_empty() {
Some(UpgradeCommand::ListProjects)
return Some(UpgradeCommand::ListProjects);
}
let target = rest.split_whitespace().next().unwrap_or(rest);
if target.eq_ignore_ascii_case("all") {
Some(UpgradeCommand::UpgradeAll)
} else {
Some(UpgradeCommand::Upgrade {
project: rest.split_whitespace().next().unwrap_or(rest).to_string(),
project: target.to_string(),
})
}
}
@@ -98,6 +114,84 @@ pub async fn handle_upgrade_list_projects(
format!("Registered projects (use `upgrade <project>` to upgrade one):\n{list}")
}
/// Resolve the artifact source URL and expected git hash for an upgrade.
///
/// The URL points at the gateway's own artifact endpoint via
/// `host.docker.internal` (resolvable from inside sled containers). The
/// expected hash comes from the `.hash` sidecar written by `release`, when
/// present. `HUSKIES_GATEWAY_BINARY_URL` overrides the URL (no hash check).
///
/// Returns `Err` with a user-facing message when no artifact has been
/// published yet.
fn resolve_artifact_source(gateway_port: Option<u16>) -> Result<(String, Option<String>), String> {
if let Ok(url) = std::env::var("HUSKIES_GATEWAY_BINARY_URL") {
return Ok((url, None));
}
let artifact_name = crate::http::SLED_ARTIFACT_NAME;
let artifact_path = crate::http::artifacts_dir().join(artifact_name);
if !artifact_path.exists() {
return Err(format!(
"No published artifact at `{}`. Run `release` first to build and publish one.",
artifact_path.display()
));
}
let expected_hash =
std::fs::read_to_string(crate::http::artifacts_dir().join(format!("{artifact_name}.hash")))
.ok()
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty());
let url = format!(
"http://host.docker.internal:{}/api/artifacts/{artifact_name}",
gateway_port.unwrap_or(3000)
);
Ok((url, expected_hash))
}
/// Upgrade every registered sled in sequence, streaming per-sled phase markers.
///
/// Sleds with active agent processes are skipped (reported, not failed).
/// Returns a summary listing the outcome for each sled.
pub async fn handle_upgrade_all<F, Fut>(
projects_store: &Arc<RwLock<BTreeMap<String, ProjectEntry>>>,
gateway_port: Option<u16>,
send_phase: F,
) -> String
where
F: Fn(String) -> Fut,
Fut: Future<Output = ()>,
{
let names: Vec<String> = {
let projects = projects_store.read().await;
projects.keys().cloned().collect()
};
if names.is_empty() {
return "No projects are currently registered with the gateway.".to_string();
}
// Fail fast before touching any sled if there is nothing to distribute.
if let Err(e) = resolve_artifact_source(gateway_port) {
return e;
}
let mut results: Vec<String> = Vec::with_capacity(names.len());
for name in &names {
let outcome = handle_sled_upgrade(name, projects_store, gateway_port, |msg| {
send_phase(format!("**{name}** {msg}"))
})
.await;
results.push(format!("- {name}: {outcome}"));
}
format!(
"Upgrade sweep over {} sled(s) complete:\n{}",
names.len(),
results.join("\n")
)
}
/// Upgrade a named sled by streaming phase markers to the chat room.
///
/// Acquires the global upgrade lock to serialise concurrent invocations. Each
@@ -105,9 +199,8 @@ pub async fn handle_upgrade_list_projects(
/// begins. On any failure, an error message is returned and the previous
/// binary remains active on the sled.
///
/// `gateway_port` is used to derive the default binary source URL
/// (`http://gateway:<port>/api/huskies-binary`) when neither
/// `HUSKIES_GATEWAY_BINARY_URL` nor `--source` is set.
/// Sleds with active agent processes are skipped so an upgrade never kills
/// in-flight work.
pub async fn handle_sled_upgrade<F, Fut>(
project: &str,
projects_store: &Arc<RwLock<BTreeMap<String, ProjectEntry>>>,
@@ -137,13 +230,41 @@ where
}
};
// ── Resolve binary source URL ────────────────────────────────────────────
let source_url = std::env::var("HUSKIES_GATEWAY_BINARY_URL").unwrap_or_else(|_| {
format!(
"http://gateway:{}/api/huskies-binary",
gateway_port.unwrap_or(3000)
)
});
// ── Resolve binary source ────────────────────────────────────────────────
let (source_url, expected_hash) = match resolve_artifact_source(gateway_port) {
Ok(v) => v,
Err(e) => return e,
};
// ── Drain check ──────────────────────────────────────────────────────────
// Never kill in-flight agent work; the caller can retry once idle.
let container_name = format!("huskies-{project}");
if let Ok(n) = super::project_rebuild::count_active_claude_processes(&container_name).await
&& n > 0
{
return format!(
"skipped — {n} active agent process(es) in `{container_name}`. Retry when idle."
);
}
run_sled_upgrade(project, &sled_url, &source_url, expected_hash, send_phase).await
}
/// Run the four-phase upgrade against a sled whose source URL is already
/// resolved. Split from [`handle_sled_upgrade`] so tests can drive the wire
/// behaviour without a published artifact on the host.
async fn run_sled_upgrade<F, Fut>(
project: &str,
sled_url: &str,
source_url: &str,
expected_hash: Option<String>,
send_phase: F,
) -> String
where
F: Fn(String) -> Fut,
Fut: Future<Output = ()>,
{
let container_name = format!("huskies-{project}");
// ── Acquire serial lock ──────────────────────────────────────────────────
let _lock = upgrade_lock().lock().await;
@@ -206,9 +327,25 @@ where
// ── Phase 4: reconnected ─────────────────────────────────────────────────
send_phase("[4/4] reconnected to gateway".to_string()).await;
// ── Report new version ───────────────────────────────────────────────────
let version = fetch_sled_version(&client, &sled_url).await;
format!("{project} upgraded to version {version}")
// ── Verify convergence ───────────────────────────────────────────────────
match fetch_sled_version(&client, sled_url).await {
Some((version, git_hash)) => match expected_hash {
Some(expected) if git_hash == expected => {
format!("upgraded to v{version} ({git_hash}) — matches published artifact")
}
Some(expected) => format!(
"**upgrade did not converge**: sled reports {git_hash}, published artifact \
is {expected}. The sled is healthy but running the wrong binary — check \
`docker logs {container_name}`."
),
None => format!("upgraded to v{version} ({git_hash})"),
},
None => format!(
"upgraded and healthy, but `/api/version` is unavailable — the sled is \
probably still on a pre-version-endpoint binary. Check `docker logs \
{container_name}`."
),
}
}
// ── Helpers ───────────────────────────────────────────────────────────────────
@@ -231,41 +368,16 @@ async fn wait_for_health(client: &reqwest::Client, health_url: &str, timeout_sec
}
}
/// Fetch the running version from the sled's `get_version` MCP tool.
/// Fetch `(version, git_hash)` from the sled's `/api/version` endpoint.
///
/// Returns the version string on success, or `"unknown"` on any error so the
/// final chat reply is still meaningful.
async fn fetch_sled_version(client: &reqwest::Client, sled_url: &str) -> String {
let mcp_url = format!("{}/mcp", sled_url.trim_end_matches('/'));
let body = serde_json::json!({
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "get_version",
"arguments": {}
}
});
let resp = match client.post(&mcp_url).json(&body).send().await {
Ok(r) => r,
Err(_) => return "unknown".to_string(),
};
let val: serde_json::Value = match resp.json().await {
Ok(v) => v,
Err(_) => return "unknown".to_string(),
};
// MCP tools/call response: result.content[0].text is a JSON string.
let text = val
.pointer("/result/content/0/text")
.and_then(|v| v.as_str())
.unwrap_or("");
if text.is_empty() {
return "unknown".to_string();
}
serde_json::from_str::<serde_json::Value>(text)
.ok()
.and_then(|v| v.get("version").and_then(|v| v.as_str()).map(String::from))
.unwrap_or_else(|| "unknown".to_string())
/// Returns `None` when the endpoint is unreachable or malformed — e.g. a sled
/// still running a binary that predates the endpoint.
async fn fetch_sled_version(client: &reqwest::Client, sled_url: &str) -> Option<(String, String)> {
let url = format!("{}/api/version", sled_url.trim_end_matches('/'));
let val: serde_json::Value = client.get(&url).send().await.ok()?.json().await.ok()?;
let version = val.get("version").and_then(|v| v.as_str())?.to_string();
let git_hash = val.get("git_hash").and_then(|v| v.as_str())?.to_string();
Some((version, git_hash))
}
// ── Tests ──────────────────────────────────────────────────────────────────────
@@ -405,24 +517,38 @@ mod tests {
);
}
#[test]
fn extract_upgrade_all() {
let cmd = extract_upgrade_command("Timmy upgrade all", "Timmy", "@timmy:home");
assert_eq!(cmd, Some(UpgradeCommand::UpgradeAll));
let cmd = extract_upgrade_command("@timmy upgrade ALL", "Timmy", "@timmy:home");
assert_eq!(cmd, Some(UpgradeCommand::UpgradeAll));
}
#[tokio::test]
async fn upgrade_all_empty_store_reports_no_projects() {
let store: Arc<RwLock<BTreeMap<String, ProjectEntry>>> =
Arc::new(RwLock::new(BTreeMap::new()));
let msg = handle_upgrade_all(&store, Some(3000), |_msg| async {}).await;
assert!(
msg.contains("No projects"),
"empty store should say no projects: {msg}"
);
}
#[tokio::test]
async fn upgrade_unreachable_sled_reports_failure() {
let mut map = BTreeMap::new();
map.insert(
"myapp".to_string(),
ProjectEntry {
url: Some("http://127.0.0.1:1".into()), // port 1 is never listening
auth_token: None,
ssh_port: None,
host_path: None,
},
);
let store = Arc::new(RwLock::new(map));
let phases: std::sync::Mutex<Vec<String>> = std::sync::Mutex::new(vec![]);
let result = handle_sled_upgrade("myapp", &store, Some(3000), |msg| {
phases.lock().unwrap().push(msg);
async {}
})
let result = run_sled_upgrade(
"myapp",
"http://127.0.0.1:1", // port 1 is never listening
"http://127.0.0.1:1/api/artifacts/huskies-linux-arm64",
None,
|msg| {
phases.lock().unwrap().push(msg);
async {}
},
)
.await;
// Phase 1 marker must have been sent before the failed request.
let sent = phases.lock().unwrap().clone();
server/src/http/mod.rs
+7
View File
@@ -298,6 +298,13 @@ pub async fn serve_binary_handler() -> poem::Response {
}
}
/// Canonical artifact filename for sled binaries on this deployment's platform.
///
/// Sleds run linux/arm64 under OrbStack on Apple Silicon. When amd64 hosts
/// arrive, `release` publishes one artifact per platform and this becomes a
/// lookup instead of a constant.
pub const SLED_ARTIFACT_NAME: &str = "huskies-linux-arm64";
/// Directory where the gateway stores distributable sled binaries.
///
/// Host-global (`~/.huskies/artifacts/`), not per-project: one artifact serves