fix(orchestrator): user stop aborts host-port readiness/repair waits

wait_for_manifest_host_ports polls up to 420s while the caller holds
the per-app lock, so a user's stop request queued behind it and the
scanner painted the app Restarting for the whole wait (the dd3afbba
transitional-state stick; gate take-2 iter 5 vaultwarden). The wait
now selects against a user-stop-marker watcher (2s poll on
user-stopped.json, by app id or container name) and aborts with a
recognizable error, and the host-port repair path skips its restart
when a stop was requested while it settled — restarting there would
fight the queued stop worker.

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

core/archipelago/src/container/prod_orchestrator.rs

@@ -493,12 +493,32 @@ async fn http_host_port_ready(port: u16, path: &str) -> bool {
         || head.starts_with("HTTP/1.0 3")
 }
 
-async fn wait_for_manifest_host_ports(manifest: &AppManifest, timeout_secs: u64) -> Result<()> {
+/// Resolves when a user-stop marker for the app (by app id or container
+/// name) appears. Long readiness/repair waits select against this so a
+/// user's stop request aborts the wait instead of queueing behind it for
+/// minutes while the app is painted Restarting (§C follow-up from the
+/// dd3afbba transitional-state stick; gate take-2 iter 5 vaultwarden).
+async fn wait_for_user_stop_marker(data_dir: &Path, app_id: &str, container_name: &str) {
+    loop {
+        let user_stopped = crate::crash_recovery::load_user_stopped(data_dir).await;
+        if user_stopped.contains(app_id) || user_stopped.contains(container_name) {
+            return;
+        }
+        tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+    }
+}
+
+async fn wait_for_manifest_host_ports(
+    manifest: &AppManifest,
+    timeout_secs: u64,
+    data_dir: &Path,
+) -> Result<()> {
     // Only TCP host ports are reachability-probed: the probe is a TCP connect,
     // which a UDP/SCTP listener (e.g. netbird's 3478/udp STUN) can never answer,
     // so probing it would always "fail" and drive an endless host-port repair
     // loop (observed on .228 after netbird's manifest deploy). Default protocol
     // (empty) is tcp.
+    let container_name = compute_container_name(manifest);
     for port in manifest
         .app
         .ports
@@ -506,9 +526,19 @@ async fn wait_for_manifest_host_ports(manifest: &AppManifest, timeout_secs: u64)
         .filter(|p| matches!(p.protocol.to_ascii_lowercase().as_str(), "" | "tcp"))
         .map(|p| p.host)
     {
-        let ready = match manifest.app.id.as_str() {
+        let ready = tokio::select! {
-            "uptime-kuma" => wait_for_http_host_port(port, "/", timeout_secs).await,
+            ready = async {
-            _ => wait_for_host_port(port, timeout_secs).await,
+                match manifest.app.id.as_str() {
+                    "uptime-kuma" => wait_for_http_host_port(port, "/", timeout_secs).await,
+                    _ => wait_for_host_port(port, timeout_secs).await,
+                }
+            } => ready,
+            _ = wait_for_user_stop_marker(data_dir, &manifest.app.id, &container_name) => {
+                return Err(anyhow::anyhow!(
+                    "aborting host-port wait for {}: user stop requested mid-wait",
+                    manifest.app.id
+                ));
+            }
         };
         if !ready {
             return Err(anyhow::anyhow!(
@@ -793,9 +823,25 @@ async fn repair_manifest_host_ports_after_stability(
     runtime: &dyn ContainerRuntimeTrait,
     manifest: &AppManifest,
     name: &str,
+    data_dir: &Path,
 ) -> Result<()> {
     tokio::time::sleep(std::time::Duration::from_secs(5)).await;
-    if wait_for_manifest_host_ports(manifest, 5).await.is_ok() {
+    if wait_for_manifest_host_ports(manifest, 5, data_dir)
+        .await
+        .is_ok()
+    {
+        return Ok(());
+    }
+
+    // A stop the user requested while we were settling wins over the repair:
+    // restarting here would immediately fight the queued stop worker.
+    let user_stopped = crate::crash_recovery::load_user_stopped(data_dir).await;
+    if user_stopped.contains(&manifest.app.id) || user_stopped.contains(name) {
+        tracing::info!(
+            app_id = %manifest.app.id,
+            container = %name,
+            "skipping host-port repair: user stop requested"
+        );
         return Ok(());
     }
 
@@ -822,7 +868,7 @@ async fn repair_manifest_host_ports_after_stability(
             .with_context(|| format!("restart container {name}"))?;
     }
 
-    wait_for_manifest_host_ports(manifest, host_port_wait_timeout_secs(manifest)).await?;
+    wait_for_manifest_host_ports(manifest, host_port_wait_timeout_secs(manifest), data_dir).await?;
     wait_for_container_stable_running(runtime, name, 15, 90).await
 }
 
@@ -1825,6 +1871,7 @@ impl ProdContainerOrchestrator {
                             if let Err(err) = wait_for_manifest_host_ports(
                                 &resolved_manifest,
                                 host_port_repair_probe_timeout_secs(&resolved_manifest),
+                                &self.data_dir,
                             )
                             .await
                             {
@@ -1833,6 +1880,7 @@ impl ProdContainerOrchestrator {
                                     self.runtime.as_ref(),
                                     &resolved_manifest,
                                     &name,
+                                    &self.data_dir,
                                 )
                                 .await?;
                                 return Ok(ReconcileAction::Started);
@@ -1900,6 +1948,7 @@ impl ProdContainerOrchestrator {
                         wait_for_manifest_host_ports(
                             &resolved_manifest,
                             host_port_wait_timeout_secs(&resolved_manifest),
+                            &self.data_dir,
                         )
                         .await?;
                         if uses_pasta_network(&resolved_manifest) {
@@ -1955,6 +2004,7 @@ impl ProdContainerOrchestrator {
                         wait_for_manifest_host_ports(
                             &resolved_manifest,
                             host_port_wait_timeout_secs(&resolved_manifest),
+                            &self.data_dir,
                         )
                         .await?;
                         if uses_pasta_network(&resolved_manifest) {
@@ -1990,6 +2040,7 @@ impl ProdContainerOrchestrator {
                     wait_for_manifest_host_ports(
                         &resolved_manifest,
                         host_port_wait_timeout_secs(&resolved_manifest),
+                        &self.data_dir,
                     )
                     .await?;
                     if uses_pasta_network(&resolved_manifest) {
@@ -2183,6 +2234,7 @@ impl ProdContainerOrchestrator {
             if let Err(err) = wait_for_manifest_host_ports(
                 &resolved_manifest,
                 host_port_wait_timeout_secs(&resolved_manifest),
+                &self.data_dir,
             )
             .await
             {
@@ -2196,6 +2248,7 @@ impl ProdContainerOrchestrator {
                     self.runtime.as_ref(),
                     &resolved_manifest,
                     &name,
+                    &self.data_dir,
                 )
                 .await?;
                 return Ok(());
@@ -2205,12 +2258,14 @@ impl ProdContainerOrchestrator {
                 self.runtime.as_ref(),
                 &resolved_manifest,
                 &name,
+                &self.data_dir,
             )
             .await?;
         } else {
             wait_for_manifest_host_ports(
                 &resolved_manifest,
                 host_port_wait_timeout_secs(&resolved_manifest),
+                &self.data_dir,
             )
             .await?;
         }
@@ -3845,6 +3900,7 @@ impl ContainerOrchestrator for ProdContainerOrchestrator {
             wait_for_manifest_host_ports(
                 &resolved_manifest,
                 host_port_wait_timeout_secs(&resolved_manifest),
+                &self.data_dir,
             )
             .await?;
             return Ok(());
@@ -3860,6 +3916,7 @@ impl ContainerOrchestrator for ProdContainerOrchestrator {
         wait_for_manifest_host_ports(
             &resolved_manifest,
             host_port_wait_timeout_secs(&resolved_manifest),
+            &self.data_dir,
         )
         .await?;
         if uses_pasta_network(&resolved_manifest) {
@@ -3868,6 +3925,7 @@ impl ContainerOrchestrator for ProdContainerOrchestrator {
                 self.runtime.as_ref(),
                 &resolved_manifest,
                 &name,
+                &self.data_dir,
             )
             .await?;
         }
@@ -4716,6 +4774,29 @@ app:
         assert!(!calls.iter().any(|c| c.starts_with("build_image:")));
     }
 
+    #[tokio::test]
+    async fn host_port_wait_aborts_on_user_stop_marker() {
+        // A stop the user requests mid-wait must break the (up to 420s)
+        // readiness poll instead of holding the app lock — the queued stop
+        // worker paints the app Restarting for the whole wait otherwise
+        // (gate take-2 iter 5 vaultwarden).
+        let yaml = "app:\n  id: stopme\n  name: stopme\n  version: 1.0.0\n  container:\n    image: stopme:1\n  ports:\n    - host: 1\n      container: 80\n      protocol: tcp\n";
+        let manifest = AppManifest::parse(yaml).unwrap();
+        let data_dir = tempfile::tempdir().unwrap();
+        crate::crash_recovery::mark_user_stopped(data_dir.path(), "stopme").await;
+
+        // Host port 1 is never listening, so only the marker can end this
+        // before the 60s wait budget; the 10s bound proves the abort fired.
+        let res = tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            wait_for_manifest_host_ports(&manifest, 60, data_dir.path()),
+        )
+        .await
+        .expect("wait must abort well before its 60s budget");
+        let err = res.unwrap_err().to_string();
+        assert!(err.contains("user stop requested"), "err: {err}");
+    }
+
     #[tokio::test]
     async fn install_fresh_skips_pull_when_image_local() {
         // An unreachable registry must not brick a reconcile recreate whose