fix(install): never fail a first-time LND install over transient conditions

Two confirmed fresh-node failure modes made 'install LND' fail on the
first try and succeed on the second:

- The dependency gate failed fast when Bitcoin was still mid-install
  (image pulling, container not created yet): DepProbe.existing only
  covered 'podman ps -a', so an in-flight dependency looked identical to
  'not installed at all'. The probe now also reports packages in
  Installing/Updating state, and the gate waits up to 30 minutes for an
  in-flight dependency install to finish (surfacing 'Waiting for X to
  finish installing…' on the card) before its normal 3-minute
  started-but-not-running window.

- Image pulls had a hard 300s per-URL wall clock, which killed
  slow-but-progressing pulls of large images (LND on fresh-node WiFi);
  the retry then succeeded off the warm layer cache. Pulls are now
  stall-aware: killed only after 3 minutes with no podman output AND no
  byte movement in the staging dir, with a 30-minute absolute ceiling.
  Dead registries still fail in ~3 minutes.

Adds dep-gate unit tests for the in-flight-install wait and the
neither-installed-nor-installing fail-fast.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

core/archipelago/src/api/rpc/package/dependencies.rs

@@ -246,6 +246,11 @@ pub(super) fn check_install_deps(package_id: &str, deps: &RunningDeps) -> Result
 pub(super) const DEP_WAIT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(5);
 /// 36 × 5s = 3 minutes of bounded waiting.
 pub(super) const DEP_WAIT_MAX_ATTEMPTS: u32 = 36;
+/// Separate, much larger budget while a dependency is still INSTALLING
+/// (image pulling, container not created yet): 360 × 5s = 30 minutes. A
+/// fresh-node Bitcoin pull routinely takes >3 minutes, and rejecting LND
+/// mid-pull was the confirmed "first install fails, second works" failure.
+pub(super) const DEP_INSTALLING_WAIT_MAX_ATTEMPTS: u32 = 360;
 
 /// Marker error: the install was rejected by the dependency gate BEFORE any
 /// resource (container, image, data dir) was created for the package. The
@@ -317,8 +322,12 @@ pub(super) struct DepProbe {
     /// Which dependency services are currently Running.
     pub running: RunningDeps,
     /// Container/package names that EXIST in any state — installed, but
-    /// possibly not running yet (`podman ps -a` ∪ package-state entries).
+    /// possibly not running yet (`podman ps -a`).
     pub existing: Vec<String>,
+    /// Package ids currently mid-install (state Installing/Updating) —
+    /// no container exists yet, but one is on the way, so the gate must
+    /// wait for the install to finish instead of failing fast.
+    pub installing: Vec<String>,
 }
 
 /// All container names known to podman in any state (`podman ps -a`).
@@ -366,8 +375,13 @@ where
     LF: std::future::Future<Output = ()>,
 {
     let mut waited_attempts = 0u32;
+    let mut installing_attempts = 0u32;
     loop {
-        let DepProbe { running, existing } = probe().await?;
+        let DepProbe {
+            running,
+            existing,
+            installing,
+        } = probe().await?;
         let missing = missing_install_deps(package_id, &running);
         if missing.is_empty() {
             // Keep behavior in lockstep with the canonical gate (covers any
@@ -377,11 +391,12 @@ where
         }
 
         // Fail fast if any missing dependency has no installed container
-        // under any name variant — waiting cannot satisfy it.
+        // under any name variant AND no install in flight — waiting cannot
+        // satisfy it.
         let some_dep_not_installed = missing.iter().any(|dep| {
-            !dep.containers
+            !dep.containers.iter().any(|c| {
-                .iter()
+                existing.iter().any(|e| e == c) || installing.iter().any(|i| i == c)
-                .any(|c| existing.iter().any(|e| e == c))
+            })
         });
         if some_dep_not_installed {
             let msg = match check_install_deps(package_id, &running) {
@@ -391,8 +406,38 @@ where
             return Err(anyhow::Error::new(DependencyGateError(msg)));
         }
 
+        // A dependency still mid-install (no container yet) gets its own,
+        // much larger budget: a fresh-node image pull can take far longer
+        // than the started-but-not-running window.
+        let some_dep_installing = missing.iter().any(|dep| {
+            dep.containers
+                .iter()
+                .any(|c| installing.iter().any(|i| i == c))
+        });
+
+        let labels = join_dep_labels(&missing);
+        if some_dep_installing {
+            if installing_attempts >= DEP_INSTALLING_WAIT_MAX_ATTEMPTS {
+                return Err(anyhow::Error::new(DependencyGateError(format!(
+                    "{labels} is still installing after {} seconds. Wait for it \
+                     to finish, then install {package_id} again.",
+                    u64::from(DEP_INSTALLING_WAIT_MAX_ATTEMPTS) * interval.as_secs()
+                ))));
+            }
+            installing_attempts += 1;
+            if installing_attempts == 1 {
+                info!(
+                    "Install {package_id}: dependency {labels} is still installing — \
+                     waiting up to {}s for it to finish",
+                    u64::from(DEP_INSTALLING_WAIT_MAX_ATTEMPTS) * interval.as_secs()
+                );
+            }
+            on_waiting(format!("Waiting for {labels} to finish installing…")).await;
+            tokio::time::sleep(interval).await;
+            continue;
+        }
+
         if waited_attempts >= max_attempts {
-            let labels = join_dep_labels(&missing);
             return Err(anyhow::Error::new(DependencyGateError(format!(
                 "{labels} is installed but did not reach the running state within \
                  {} seconds. Start {labels}, then install {package_id} again.",
@@ -401,7 +446,6 @@ where
         }
         waited_attempts += 1;
 
-        let labels = join_dep_labels(&missing);
         if waited_attempts == 1 {
             info!(
                 "Install {package_id}: dependency {labels} installed but not running yet — \
@@ -874,9 +918,19 @@ mod tests {
         }
 
         fn probe(has_bitcoin: bool, has_electrumx: bool, existing: &[&str]) -> DepProbe {
+            probe_with_installing(has_bitcoin, has_electrumx, existing, &[])
+        }
+
+        fn probe_with_installing(
+            has_bitcoin: bool,
+            has_electrumx: bool,
+            existing: &[&str],
+            installing: &[&str],
+        ) -> DepProbe {
             DepProbe {
                 running: deps(has_bitcoin, has_electrumx),
                 existing: existing.iter().map(|s| s.to_string()).collect(),
+                installing: installing.iter().map(|s| s.to_string()).collect(),
             }
         }
 
@@ -965,6 +1019,71 @@ mod tests {
             assert!(labels.iter().all(|l| l == "Waiting for Bitcoin to start…"));
         }
 
+        #[tokio::test]
+        async fn waits_while_dependency_is_still_installing_then_passes() {
+            // Bitcoin has NO container yet (image still pulling) but its
+            // package state is Installing. Old behavior failed fast here —
+            // the confirmed fresh-node LND "first install fails" bug. The
+            // gate must wait past the normal started-but-not-running budget
+            // (max_attempts=1 below) while the install is in flight.
+            let calls = Arc::new(AtomicU32::new(0));
+            let (labels, sink) = label_sink();
+            let probe_calls = Arc::clone(&calls);
+            let result = wait_for_install_deps(
+                "lnd",
+                move || {
+                    let n = probe_calls.fetch_add(1, Ordering::SeqCst);
+                    async move {
+                        Ok(match n {
+                            // pulling: no container, package Installing
+                            0 | 1 => probe_with_installing(false, false, &[], &["bitcoin-knots"]),
+                            // container created, starting
+                            2 => probe(false, false, &["bitcoin-knots"]),
+                            // running
+                            _ => probe(true, false, &["bitcoin-knots"]),
+                        })
+                    }
+                },
+                sink,
+                1,
+                Duration::ZERO,
+            )
+            .await;
+            assert!(result.is_ok(), "{result:?}");
+            assert_eq!(calls.load(Ordering::SeqCst), 4);
+            let labels = labels.lock().unwrap();
+            assert_eq!(
+                labels.as_slice(),
+                [
+                    "Waiting for Bitcoin to finish installing…",
+                    "Waiting for Bitcoin to finish installing…",
+                    "Waiting for Bitcoin to start…",
+                ]
+            );
+        }
+
+        #[tokio::test]
+        async fn fails_fast_when_dependency_neither_installed_nor_installing() {
+            // installing list has unrelated packages only — no reason to wait.
+            let calls = AtomicU32::new(0);
+            let (labels, sink) = label_sink();
+            let err = wait_for_install_deps(
+                "lnd",
+                || {
+                    calls.fetch_add(1, Ordering::SeqCst);
+                    async { Ok(probe_with_installing(false, false, &[], &["grafana"])) }
+                },
+                sink,
+                36,
+                Duration::ZERO,
+            )
+            .await
+            .unwrap_err();
+            assert_eq!(calls.load(Ordering::SeqCst), 1);
+            assert!(labels.lock().unwrap().is_empty());
+            assert!(err.downcast_ref::<DependencyGateError>().is_some());
+        }
+
         #[tokio::test]
         async fn times_out_when_installed_dependency_never_runs() {
             let (labels, sink) = label_sink();

core/archipelago/src/api/rpc/package/install.rs

@@ -984,6 +984,24 @@ impl RpcHandler {
         }))
     }
 
+    /// Package ids currently mid-install/update per the state manager — used
+    /// by the dependency gate so an app whose dependency is still pulling its
+    /// image (no container yet) waits instead of failing fast.
+    async fn packages_currently_installing(&self) -> Vec<String> {
+        let (data, _) = self.state_manager.get_snapshot().await;
+        data.package_data
+            .iter()
+            .filter(|(_, entry)| {
+                matches!(
+                    entry.state,
+                    crate::data_model::PackageState::Installing
+                        | crate::data_model::PackageState::Updating
+                )
+            })
+            .map(|(id, _)| id.clone())
+            .collect()
+    }
+
     async fn running_deps_for_install(&self, package_id: &str) -> Result<RunningDeps> {
         let (data, _) = self.state_manager.get_snapshot().await;
         let cached = detect_running_deps_from_package_data(&data.package_data);
@@ -1006,6 +1024,7 @@ impl RpcHandler {
                 Ok(DepProbe {
                     running: self.running_deps_for_install(package_id).await?,
                     existing: detect_existing_containers().await,
+                    installing: self.packages_currently_installing().await,
                 })
             },
             |msg| async move { self.set_install_message(package_id, &msg).await },
@@ -1089,46 +1108,97 @@ impl RpcHandler {
             .spawn()
             .context("Failed to start image pull")?;
 
-        // 5-minute per-URL budget. A full install tries each configured mirror
+        // Stall-aware budget instead of a hard wall clock. The old fixed
-        // once, so a two-registry setup fails visibly in roughly 10 minutes
+        // 300s cap killed slow-but-progressing pulls (LND on fresh-node
-        // instead of staying in Installing for up to an hour.
+        // WiFi routinely needs longer), which surfaced as "first install
-        const PULL_URL_TIMEOUT_SECS: u64 = 300;
+        // fails, second succeeds" once the layer cache was warm. A pull is
-        let pull_result = tokio::time::timeout(
+        // only killed when NOTHING is observably happening — no stderr
-            std::time::Duration::from_secs(PULL_URL_TIMEOUT_SECS),
+        // output and no byte growth in podman's staging dir — for
-            async {
+        // PULL_STALL_TIMEOUT_SECS, or at a generous absolute ceiling.
-                if let Some(stderr) = child.stderr.take() {
+        // Dead registries still fail fast: no bytes ever land, so the
-                    let reader = BufReader::new(stderr);
+        // stall window (3 min) is the effective bound.
-                    let mut lines = reader.lines();
+        const PULL_STALL_TIMEOUT_SECS: u64 = 180;
-                    let pkg_id = package_id.to_string();
+        const PULL_URL_MAX_SECS: u64 = 1800;
-                    let state_mgr = self.state_manager.clone();
+        const PULL_POLL_INTERVAL_SECS: u64 = 5;
 
-                    while let Ok(Some(line)) = lines.next_line().await {
+        let started = std::time::Instant::now();
-                        if let Some((downloaded, total)) = parse_pull_progress(&line) {
+        // Seconds-since-start of the last stderr line, updated by the
-                            Self::update_install_progress(&state_mgr, &pkg_id, downloaded, total)
+        // reader task. u64::MAX sentinel = no line seen yet (treated as
-                                .await;
+        // activity at t=0 so the stall window starts at spawn).
-                        }
+        let last_line_at = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
+        if let Some(stderr) = child.stderr.take() {
+            let reader = BufReader::new(stderr);
+            let mut lines = reader.lines();
+            let pkg_id = package_id.to_string();
+            let state_mgr = self.state_manager.clone();
+            let line_clock = std::sync::Arc::clone(&last_line_at);
+            let started_reader = started;
+            tokio::spawn(async move {
+                while let Ok(Some(line)) = lines.next_line().await {
+                    line_clock.store(
+                        started_reader.elapsed().as_secs(),
+                        std::sync::atomic::Ordering::Relaxed,
+                    );
+                    if let Some((downloaded, total)) = parse_pull_progress(&line) {
+                        Self::update_install_progress(&state_mgr, &pkg_id, downloaded, total)
+                            .await;
                     }
                 }
-                child.wait().await
+            });
-            },
+        }
-        )
-        .await;
 
-        match pull_result {
+        let mut last_staged_bytes = dir_size_bytes(user_tmp);
-            Ok(Ok(status)) => Ok(status.success()),
+        let mut last_staged_change = std::time::Instant::now();
-            Ok(Err(e)) => {
+        loop {
-                tracing::warn!("Image pull process error on {}: {}", url, e);
+            match child.try_wait() {
-                Ok(false)
+                Ok(Some(status)) => return Ok(status.success()),
+                Ok(None) => {}
+                Err(e) => {
+                    tracing::warn!("Image pull process error on {}: {}", url, e);
+                    let _ = child.kill().await;
+                    let _ = child.wait().await;
+                    return Ok(false);
+                }
             }
-            Err(_) => {
+
+            tokio::time::sleep(std::time::Duration::from_secs(PULL_POLL_INTERVAL_SECS)).await;
+
+            if started.elapsed() > std::time::Duration::from_secs(PULL_URL_MAX_SECS) {
                 tracing::warn!(
-                    "Image pull timed out after {}s: {}",
+                    "Image pull exceeded absolute {}s ceiling: {}",
-                    PULL_URL_TIMEOUT_SECS,
+                    PULL_URL_MAX_SECS,
                     url
                 );
                 let _ = child.kill().await;
                 let _ = child.wait().await; // reap zombie
-                Ok(false)
+                return Ok(false);
+            }
+
+            // Activity signal 1: stderr output from podman.
+            let line_age = started
+                .elapsed()
+                .as_secs()
+                .saturating_sub(last_line_at.load(std::sync::atomic::Ordering::Relaxed));
+            // Activity signal 2: podman stages layer downloads in TMPDIR
+            // (user_tmp) — any size change there means bytes are moving.
+            let staged = dir_size_bytes(user_tmp);
+            if staged != last_staged_bytes {
+                last_staged_bytes = staged;
+                last_staged_change = std::time::Instant::now();
+            }
+
+            let stalled = line_age > PULL_STALL_TIMEOUT_SECS
+                && last_staged_change.elapsed()
+                    > std::time::Duration::from_secs(PULL_STALL_TIMEOUT_SECS);
+            if stalled {
+                tracing::warn!(
+                    "Image pull stalled ({}s with no output and no staged bytes): {}",
+                    PULL_STALL_TIMEOUT_SECS,
+                    url
+                );
+                let _ = child.kill().await;
+                let _ = child.wait().await; // reap zombie
+                return Ok(false);
             }
         }
     }
@@ -2690,6 +2760,31 @@ async fn persist_install_version_selection(app_id: &str, version: &str) {
     }
 }
 
+/// Total bytes under `path` (bounded recursive walk). Used as a coarse
+/// "bytes are moving" signal for pull stall detection — exact size doesn't
+/// matter, only whether it CHANGES between polls. Errors count as 0.
+fn dir_size_bytes(path: &str) -> u64 {
+    fn walk(dir: &std::path::Path, depth: u32) -> u64 {
+        if depth > 8 {
+            return 0;
+        }
+        let Ok(entries) = std::fs::read_dir(dir) else {
+            return 0;
+        };
+        let mut total = 0u64;
+        for entry in entries.flatten() {
+            let Ok(meta) = entry.metadata() else { continue };
+            if meta.is_dir() {
+                total = total.saturating_add(walk(&entry.path(), depth + 1));
+            } else {
+                total = total.saturating_add(meta.len());
+            }
+        }
+        total
+    }
+    walk(std::path::Path::new(path), 0)
+}
+
 fn should_try_orchestrator_install(package_id: &str, orchestrator_available: bool) -> bool {
     orchestrator_available && uses_orchestrator_install_flow(package_id)
 }