fix(proxies): stop the tunnel before the earner, so a stop that fails halfway is not undone

app/proxy_tunnel.py

@@ -435,19 +435,24 @@ def _is_ours(container: Any | None, slug: str, proxy_id: int) -> bool:
 
 
 def stop(slug: str, proxy_id: int) -> bool:
-    """Stop the earner, then its tunnel. Returns whether either was running.
+    """Stop the tunnel, then its earner. Returns whether either was running.
 
     Docker answers a stop of a stopped container with success, so the cap asks
     this to tell a copy it stopped from one the owner had already stopped.
+
+    The tunnel goes first so a stop that fails halfway never leaves a ready
+    tunnel beside a stopped earner: the heartbeat starts an earner in that
+    state (earner_action), which would undo the stop. An earner left running
+    after its tunnel stopped has no way out, and earner_action leaves it alone.
     """
     earner, sidecar = _pair(slug, proxy_id)
     if not _is_ours(earner, slug, proxy_id) and not _is_ours(sidecar, slug, proxy_id):
         raise ValueError(f"No proxied copy of {slug} through proxy {proxy_id} on this worker")
     was_running = any(_is_ours(c, slug, proxy_id) and getattr(c, "status", "") in _LIVE for c in (earner, sidecar))
-    if _is_ours(earner, slug, proxy_id):
-        earner.stop(timeout=orchestrator._get_stop_timeout(slug))
     if _is_ours(sidecar, slug, proxy_id):
         sidecar.stop(timeout=10)
+    if _is_ours(earner, slug, proxy_id):
+        earner.stop(timeout=orchestrator._get_stop_timeout(slug))
     return was_running
 
 

tests/test_proxy_tunnel.py

@@ -700,14 +700,33 @@ def _pair(self, earner_status="running", tunnel_status="running"):
         tunnel = _container("cashpilot-tunnel-x-p1", "x", proxy=1, role="tunnel", status=tunnel_status)
         return earner, tunnel
 
-    def test_stop_takes_the_earner_before_its_tunnel(self):
+    def test_stop_takes_the_tunnel_before_its_earner(self):
         earner, tunnel = self._pair()
         order = MagicMock()
         order.attach_mock(earner.stop, "earner")
         order.attach_mock(tunnel.stop, "tunnel")
         with patch.object(orchestrator, "_get_client", return_value=_client_with(earner, tunnel)):
             proxy_tunnel.stop("x", 1)
-        assert [c[0] for c in order.mock_calls] == ["earner", "tunnel"]
+        assert [c[0] for c in order.mock_calls] == ["tunnel", "earner"]
+
+    def test_a_stop_that_fails_halfway_is_not_undone_by_the_heartbeat(self):
+        """A ready tunnel beside a stopped earner is what the heartbeat starts
+        again, so a failed stop must never leave that state."""
+        for failing in ("tunnel", "earner"):
+            earner, tunnel = self._pair()
+
+            def stopped(c):
+                return lambda **_kw: setattr(c, "status", "exited")
+
+            earner.stop.side_effect = stopped(earner)
+            tunnel.stop.side_effect = stopped(tunnel)
+            (tunnel if failing == "tunnel" else earner).stop.side_effect = RuntimeError("daemon timeout")
+            with (
+                patch.object(orchestrator, "_get_client", return_value=_client_with(earner, tunnel)),
+                pytest.raises(RuntimeError),
+            ):
+                proxy_tunnel.stop("x", 1)
+            assert proxy_tunnel.earner_action(earner, tunnel, ready=True) is None, failing
 
     def test_stop_and_start_refuse_a_copy_that_is_not_ours(self):
         stranger = _container("cashpilot-x-p1", "y", proxy=1)