lfg2025/archy
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects Releases 33 Wiki Activity

refactor(install): route orchestrator-managed apps through orchestrator first

Browse Source 
Phase 3a of the install path consolidation. Two coupled changes:

1. install.rs handle_package_install: gate the legacy "container exists →
   adopt + return" probe on !orchestrator_managed. Apps the orchestrator
   knows about (bitcoin-knots, bitcoin-core, lnd, electrumx, fedimint,
   filebrowser, btcpay-server stack apps, mempool stack apps, plus the
   companion UIs that just moved to Quadlet) skip the legacy probe and
   fall straight into the orchestrator branch.

   The legacy adopt block was returning success on a bare `podman start`
   exit-0 — even when the process inside the container crashed seconds
   later. That's the .228 "running but unreachable" failure mode. The
   orchestrator's ensure_running honors the manifest's health check and
   pre-start hooks (e.g. re-renders bitcoin-ui's nginx.conf if the RPC
   password rotated), so this is a behavioral upgrade, not just a
   refactor.

2. ProdContainerOrchestrator::install: make idempotent. Previously it
   blindly called install_fresh which would fail on `podman create` if
   the container name already existed. Now it delegates to ensure_running:
     - Container Running + healthy → no-op (refresh hooks, restart if
       config rewritten)
     - Container Stopped/Exited → start (with hook refresh)
     - Container missing → install_fresh
     - Container in wedged state (Created/Paused/Unknown) → force-recreate

   Without this, change #1 would regress every "container already exists"
   case for the 18 orchestrator-managed app IDs. With it, install becomes
   the single source of truth for "make app X be in the desired state."

Tests: 654 passed across the workspace (614 unit + 37 orchestration + 3
rpc), 0 failures. The 20 prod_orchestrator tests cover the install /
ensure_running / reconcile paths the new install delegates through.

Net delta: install.rs grows by ~30 lines (gating wrapper + comments),
prod_orchestrator.rs grows by ~30 lines (idempotent install body). Both
are temporary — the larger deletions (~1700 lines) come once every app
has been verified through the orchestrator path in subsequent phases.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
archipelago
2026-05-01 11:12:52 -04:00
parent 6e716f68b6
commit fdb1f3d4d3
6 changed files with 194 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

202
core/archipelago/src/api/rpc/package/install.rs
View File

@@ -120,113 +120,133 @@ impl RpcHandler {
false false
}; };
// Check if container already exists // For orchestrator-managed apps, skip the legacy "container exists →
let check_output = tokio::process::Command::new("podman") // adopt + return" probe entirely. The orchestrator's own install path
.args([ // (below) calls ensure_running which:
"ps", // - no-ops if the container is already up and healthy,
"-a", // - removes + reinstalls if the container is broken,
"--format", // - actually verifies health via the manifest's health check
"{{.Names}}", // (whereas the legacy adopt block returns success on a podman
"--filter", // `start` exit-0, even if the process inside crashed seconds
&format!("name=^{}$", package_id), // later — the .228 bitcoin "running but unreachable" failure
]) // mode).
.output() // The adoption block is being phased out as apps move to the
.await // orchestrator path. Non-orchestrator apps still hit it.
.context("Failed to check existing containers")?; let orchestrator_managed =
should_try_orchestrator_install(package_id, self.orchestrator.is_some());
if !String::from_utf8_lossy(&check_output.stdout) // Check if container already exists (legacy adoption — non-orchestrator
.trim() // apps only).
.is_empty() if !orchestrator_managed {
{ let check_output = tokio::process::Command::new("podman")
// Container already exists (e.g. created by first-boot) — adopt it .args([
info!( "ps",
"Container {} already exists, adopting as installed", "-a",
package_id "--format",
); "{{.Names}}",
install_log(&format!( "--filter",
"INSTALL ADOPT: {} — container already exists", &format!("name=^{}$", package_id),
package_id ])
))
.await;
// Check container state
let state_output = tokio::process::Command::new("podman")
.args(["inspect", package_id, "--format", "{{.State.Status}}"])
.output() .output()
.await .await
.context("Failed to inspect existing container")?; .context("Failed to check existing containers")?;
let state = String::from_utf8_lossy(&state_output.stdout)
.trim()
.to_string();
if state == "running" && repaired_bitcoin_conf { if !String::from_utf8_lossy(&check_output.stdout)
.trim()
.is_empty()
{
// Container already exists (e.g. created by first-boot) — adopt it
info!( info!(
"Restarting existing container {} after bitcoin.conf RPC repair", "Container {} already exists, adopting as installed",
package_id package_id
); );
let restart_output = tokio::process::Command::new("podman") install_log(&format!(
.args(["restart", package_id]) "INSTALL ADOPT: {} — container already exists",
package_id
))
.await;
// Check container state
let state_output = tokio::process::Command::new("podman")
.args(["inspect", package_id, "--format", "{{.State.Status}}"])
.output() .output()
.await .await
.context("Failed to restart existing container after bitcoin.conf repair")?; .context("Failed to inspect existing container")?;
if !restart_output.status.success() { let state = String::from_utf8_lossy(&state_output.stdout)
let stderr = String::from_utf8_lossy(&restart_output.stderr); .trim()
install_log(&format!( .to_string();
"INSTALL ADOPT FAIL: {} - restart after RPC repair failed: {}",
package_id, stderr if state == "running" && repaired_bitcoin_conf {
)) info!(
.await; "Restarting existing container {} after bitcoin.conf RPC repair",
return Err(anyhow::anyhow!( package_id
"Container {} exists but failed to restart after RPC repair: {}", );
package_id, let restart_output = tokio::process::Command::new("podman")
stderr .args(["restart", package_id])
)); .output()
} .await
let _ = tokio::process::Command::new("podman") .context(
.args(["restart", "archy-bitcoin-ui"]) "Failed to restart existing container after bitcoin.conf repair",
.output() )?;
.await; if !restart_output.status.success() {
wait_for_adopted_container(package_id, package_id).await?; let stderr = String::from_utf8_lossy(&restart_output.stderr);
} else if state != "running" { install_log(&format!(
// Start the stopped/exited container "INSTALL ADOPT FAIL: {} - restart after RPC repair failed: {}",
info!("Starting existing container {} (was {})", package_id, state); package_id, stderr
let start_output = tokio::process::Command::new("podman") ))
.args(["start", package_id]) .await;
.output() return Err(anyhow::anyhow!(
.await "Container {} exists but failed to restart after RPC repair: {}",
.context("Failed to start existing container")?; package_id,
if !start_output.status.success() { stderr
let stderr = String::from_utf8_lossy(&start_output.stderr); ));
install_log(&format!( }
"INSTALL ADOPT FAIL: {} — start failed: {}", let _ = tokio::process::Command::new("podman")
package_id, stderr .args(["restart", "archy-bitcoin-ui"])
)) .output()
.await; .await;
return Err(anyhow::anyhow!( wait_for_adopted_container(package_id, package_id).await?;
"Container {} exists but failed to start: {}", } else if state != "running" {
package_id, // Start the stopped/exited container
stderr info!("Starting existing container {} (was {})", package_id, state);
)); let start_output = tokio::process::Command::new("podman")
.args(["start", package_id])
.output()
.await
.context("Failed to start existing container")?;
if !start_output.status.success() {
let stderr = String::from_utf8_lossy(&start_output.stderr);
install_log(&format!(
"INSTALL ADOPT FAIL: {} — start failed: {}",
package_id, stderr
))
.await;
return Err(anyhow::anyhow!(
"Container {} exists but failed to start: {}",
package_id,
stderr
));
}
wait_for_adopted_container(package_id, package_id).await?;
} }
wait_for_adopted_container(package_id, package_id).await?; install_log(&format!(
"INSTALL ADOPT OK: {} — already running",
package_id
))
.await;
return Ok(serde_json::json!({
"success": true,
"package_id": package_id,
"message": format!("Package {} already installed and running", package_id)
}));
} }
install_log(&format!(
"INSTALL ADOPT OK: {} — already running",
package_id
))
.await;
return Ok(serde_json::json!({
"success": true,
"package_id": package_id,
"message": format!("Package {} already installed and running", package_id)
}));
} }
// Preferred path for apps already modeled in the production orchestrator. // Preferred path for apps already modeled in the production orchestrator.
// Keep legacy install flow as default while migration is in progress. // Keep legacy install flow as default while migration is in progress.
if should_try_orchestrator_install(package_id, self.orchestrator.is_some()) { if orchestrator_managed {
let orchestrator_app_id = orchestrator_install_app_id(package_id); let orchestrator_app_id = orchestrator_install_app_id(package_id);
self.set_install_phase(package_id, InstallPhase::CreatingContainer) self.set_install_phase(package_id, InstallPhase::CreatingContainer)
.await; .await;

9
core/archipelago/src/bitcoin_status.rs
View File

@@ -132,11 +132,10 @@ async fn fetch_bitcoin_status() -> Result<BitcoinNodeStatus> {
.await .await
.context("getindexinfo") .context("getindexinfo")
.ok(); .ok();
let zmq_notifications = let zmq_notifications = bitcoin_rpc_call(&client, "getzmqnotifications", serde_json::json!([]))
bitcoin_rpc_call(&client, "getzmqnotifications", serde_json::json!([])) .await
.await .context("getzmqnotifications")
.context("getzmqnotifications") .ok();
.ok();
Ok(BitcoinNodeStatus { Ok(BitcoinNodeStatus {
ok: true, ok: true,

16
core/archipelago/src/container/boot_reconciler.rs
View File

@@ -103,9 +103,7 @@ impl BootReconciler {
return; return;
} }
let installed = self.orchestrator.manifest_ids().await; let installed = self.orchestrator.manifest_ids().await;
for (companion, err) in for (companion, err) in crate::container::companion::reconcile(&installed).await {
crate::container::companion::reconcile(&installed).await
{
tracing::warn!( tracing::warn!(
companion = %companion, companion = %companion,
error = %err, error = %err,
@@ -260,7 +258,8 @@ mod tests {
let orch = orch_with_one_running_manifest(rt.clone()).await; let orch = orch_with_one_running_manifest(rt.clone()).await;
let shutdown = Arc::new(Notify::new()); let shutdown = Arc::new(Notify::new());
let reconciler = let reconciler =
BootReconciler::new(orch.clone(), Duration::from_secs(30), shutdown.clone()).without_companion_stage(); BootReconciler::new(orch.clone(), Duration::from_secs(30), shutdown.clone())
.without_companion_stage();
let handle = tokio::spawn(reconciler.run_forever()); let handle = tokio::spawn(reconciler.run_forever());
// Yield so the spawned task gets CPU to run its initial reconcile. // Yield so the spawned task gets CPU to run its initial reconcile.
@@ -284,7 +283,8 @@ mod tests {
let orch = orch_with_one_running_manifest(rt.clone()).await; let orch = orch_with_one_running_manifest(rt.clone()).await;
let shutdown = Arc::new(Notify::new()); let shutdown = Arc::new(Notify::new());
let reconciler = let reconciler =
BootReconciler::new(orch.clone(), Duration::from_secs(30), shutdown.clone()).without_companion_stage(); BootReconciler::new(orch.clone(), Duration::from_secs(30), shutdown.clone())
.without_companion_stage();
let handle = tokio::spawn(reconciler.run_forever()); let handle = tokio::spawn(reconciler.run_forever());
tokio::task::yield_now().await; tokio::task::yield_now().await;
@@ -313,7 +313,8 @@ mod tests {
let orch = orch_with_one_running_manifest(rt.clone()).await; let orch = orch_with_one_running_manifest(rt.clone()).await;
let shutdown = Arc::new(Notify::new()); let shutdown = Arc::new(Notify::new());
let reconciler = let reconciler =
BootReconciler::new(orch.clone(), Duration::from_secs(30), shutdown.clone()).without_companion_stage(); BootReconciler::new(orch.clone(), Duration::from_secs(30), shutdown.clone())
.without_companion_stage();
let handle = tokio::spawn(reconciler.run_forever()); let handle = tokio::spawn(reconciler.run_forever());
tokio::task::yield_now().await; tokio::task::yield_now().await;
tokio::task::yield_now().await; tokio::task::yield_now().await;
@@ -347,7 +348,8 @@ mod tests {
.await; .await;
let shutdown = Arc::new(Notify::new()); let shutdown = Arc::new(Notify::new());
let reconciler = let reconciler =
BootReconciler::new(orch.clone(), Duration::from_secs(30), shutdown.clone()).without_companion_stage(); BootReconciler::new(orch.clone(), Duration::from_secs(30), shutdown.clone())
.without_companion_stage();
let handle = tokio::spawn(reconciler.run_forever()); let handle = tokio::spawn(reconciler.run_forever());
tokio::task::yield_now().await; tokio::task::yield_now().await;

14
core/archipelago/src/container/companion.rs
View File

@@ -26,9 +26,7 @@ use tokio::fs;
use tokio::process::Command; use tokio::process::Command;
use tracing::{info, warn}; use tracing::{info, warn};
use crate::container::quadlet::{ use crate::container::quadlet::{self, BindMount, NetworkMode, QuadletUnit};
self, BindMount, NetworkMode, QuadletUnit,
};
use archipelago_container::image_uses_insecure_registry; use archipelago_container::image_uses_insecure_registry;
const COMPANION_REGISTRY: &str = "146.59.87.168:3000/lfg2025"; const COMPANION_REGISTRY: &str = "146.59.87.168:3000/lfg2025";
@@ -152,7 +150,10 @@ pub async fn remove_for(package_id: &str) {
pub async fn install_one(spec: &CompanionSpec) -> Result<()> { pub async fn install_one(spec: &CompanionSpec) -> Result<()> {
if let Some(hook) = spec.pre_start { if let Some(hook) = spec.pre_start {
hook().await.with_context(|| { hook().await.with_context(|| {
format!("pre-start hook failed for {} — companion will not start", spec.name) format!(
"pre-start hook failed for {} — companion will not start",
spec.name
)
})?; })?;
} }
let image = ensure_image_present(spec).await?; let image = ensure_image_present(spec).await?;
@@ -284,7 +285,10 @@ pub async fn reconcile(installed_apps: &[String]) -> Vec<(String, anyhow::Error)
match needs_repair(spec).await { match needs_repair(spec).await {
Ok(false) => {} Ok(false) => {}
Ok(true) => { Ok(true) => {
info!(companion = spec.name, "reconcile: companion not active, repairing"); info!(
companion = spec.name,
"reconcile: companion not active, repairing"
);
if let Err(e) = install_one(spec).await { if let Err(e) = install_one(spec).await {
failures.push((spec.name.to_string(), e)); failures.push((spec.name.to_string(), e));
} }

38
core/archipelago/src/container/prod_orchestrator.rs
View File

@@ -707,11 +707,41 @@ enum HookOutcome {
#[async_trait] #[async_trait]
impl ContainerOrchestrator for ProdContainerOrchestrator { impl ContainerOrchestrator for ProdContainerOrchestrator {
async fn install(&self, app_id: &str) -> Result<String> { async fn install(&self, app_id: &str) -> Result<String> {
// Idempotent: if the container is already up and healthy, just
// refresh hooks and return. If it's stopped, start it. If it's
// missing or in a wedged state, install fresh.
//
// The RPC install handler used to do its own "container exists →
// adopt + return" probe before calling here, but that path skipped
// health verification (the .228 "running but unreachable" failure
// mode). Routing every install through here means the orchestrator
// is the one source of truth for what "installed" means.
let lm = self.loaded(app_id).await?; let lm = self.loaded(app_id).await?;
let lock = self.app_lock(app_id).await; let name = compute_container_name(&lm.manifest);
let _guard = lock.lock().await; // ensure_running takes the per-app lock itself; release the install
self.install_fresh(&lm).await?; // path lock first if we hold one (we don't — install is the entry
Ok(compute_container_name(&lm.manifest)) // point). Just delegate.
let action = self.ensure_running(&lm).await?;
match action {
ReconcileAction::NoOp
| ReconcileAction::Started
| ReconcileAction::Installed => Ok(name),
ReconcileAction::Left(state) => {
// Container is in a wedged state (Created / Paused / Unknown).
// Force-recreate so the install RPC has a clean outcome.
tracing::warn!(
app_id = %app_id,
state = %state,
"install: container in wedged state, force-recreating"
);
let lock = self.app_lock(app_id).await;
let _guard = lock.lock().await;
let _ = self.runtime.stop_container(&name).await;
let _ = self.runtime.remove_container(&name).await;
self.install_fresh(&lm).await?;
Ok(name)
}
}
} }
async fn start(&self, app_id: &str) -> Result<()> { async fn start(&self, app_id: &str) -> Result<()> {

56
core/archipelago/src/electrs_status.rs
View File

@@ -86,11 +86,9 @@ pub fn spawn_status_cache() {
fresh.progress_pct = cached.progress_pct; fresh.progress_pct = cached.progress_pct;
} }
fresh.stale = true; fresh.stale = true;
fresh.error = Some( fresh.error = Some(fresh.error.unwrap_or_else(|| {
fresh "ElectrumX is reconnecting; showing last known indexed height.".to_string()
.error }));
.unwrap_or_else(|| "ElectrumX is reconnecting; showing last known indexed height.".to_string()),
);
} }
*cached = fresh; *cached = fresh;
drop(cached); drop(cached);
@@ -373,16 +371,16 @@ async fn fetch_electrs_sync_status() -> ElectrsSyncStatus {
Err(e) => match electrumx_log_indexed_height().await { Err(e) => match electrumx_log_indexed_height().await {
Ok(h) if h > 0 => h, Ok(h) if h > 0 => h,
_ => { _ => {
let err_msg = e.to_string(); let err_msg = e.to_string();
if is_transient_error(&err_msg) { if is_transient_error(&err_msg) {
// ElectrumX is starting up or busy — estimate from data size // ElectrumX is starting up or busy — estimate from data size
let progress_pct = if data_bytes > 0 { let progress_pct = if data_bytes > 0 {
((data_bytes as f64 / ESTIMATED_FULL_INDEX_BYTES) * 100.0).min(99.0) ((data_bytes as f64 / ESTIMATED_FULL_INDEX_BYTES) * 100.0).min(99.0)
} else { } else {
0.0 0.0
}; };
let size_str = index_size.clone().unwrap_or_else(|| "0 MB".to_string()); let size_str = index_size.clone().unwrap_or_else(|| "0 MB".to_string());
return ElectrsSyncStatus { return ElectrsSyncStatus {
indexed_height: 0, indexed_height: 0,
bitcoin_height: bitcoin_blocks, bitcoin_height: bitcoin_blocks,
network_height, network_height,
@@ -396,21 +394,21 @@ async fn fetch_electrs_sync_status() -> ElectrsSyncStatus {
index_size, index_size,
tor_onion, tor_onion,
}; };
}
// Genuine unexpected error
warn!("ElectrumX status: unexpected error: {}", err_msg);
return ElectrsSyncStatus {
indexed_height: 0,
bitcoin_height: bitcoin_blocks,
network_height,
progress_pct: 0.0,
status: "error".to_string(),
stale: false,
error: Some(format!("ElectrumX: {}", err_msg)),
index_size,
tor_onion,
};
} }
// Genuine unexpected error
warn!("ElectrumX status: unexpected error: {}", err_msg);
return ElectrsSyncStatus {
indexed_height: 0,
bitcoin_height: bitcoin_blocks,
network_height,
progress_pct: 0.0,
status: "error".to_string(),
stale: false,
error: Some(format!("ElectrumX: {}", err_msg)),
index_size,
tor_onion,
};
}
}, },
}; };