fix(isolation): wait for another process's fence helper instead of killing it

app/lan_fence.py

@@ -57,6 +57,7 @@
 import re
 import threading
 import time
+from datetime import datetime
 from pathlib import Path
 from typing import Any
 
@@ -770,58 +771,105 @@ def _helper_image(client: Any) -> str:
     return f"drumsergio/cashpilot-worker:{version.current()}"
 
 
-def _remove_helper(client: Any) -> None:
-    """Remove a leftover helper by name, only when it carries the helper's label.
+#: How long a helper that was just created or has just exited still belongs to
+#: the process that started it: that process starts it, then reads its logs and
+#: removes it, all within moments. HELPER_POLL is how often a busy helper is
+#: looked at again.
+HELPER_GRACE = 10.0
+HELPER_POLL = 0.25
 
-    A container someone else named cashpilot-fence is left alone; the run that
-    follows then fails on the name instead of deleting it.
+
+def _stamp_age(stamp: Any, now: float) -> float | None:
+    """Seconds since a Docker timestamp, or None when it is unset or unreadable."""
+    try:
+        when = datetime.fromisoformat(str(stamp))
+    except ValueError:
+        return None
+    if when.tzinfo is None or when.year < 2000:
+        return None
+    return now - when.timestamp()
+
+
+def _helper_busy(attrs: dict[str, Any], now: float) -> bool:
+    """A helper another process is still using: running, or created or exited within HELPER_GRACE."""
+    state = attrs.get("State") or {}
+    if state.get("Running") or state.get("Restarting"):
+        return True
+    status = state.get("Status")
+    if status == "created":
+        stamp = attrs.get("Created")
+    elif status in ("exited", "dead"):
+        stamp = state.get("FinishedAt")
+    else:
+        return False
+    age = _stamp_age(stamp, now)
+    return age is not None and age < HELPER_GRACE
+
+
+def _clear_helper(client: Any) -> None:
+    """Make room for this process's helper.
+
+    Only a container that carries the helper's label is touched: a container
+    someone else named cashpilot-fence is left alone, and the run that follows
+    fails on the name instead of deleting it.
+
+    The Desktop app runs the same helper, by the same name and label, so on a
+    host that runs both, a helper of ours can belong to another process that is
+    still waiting for its result. That one is waited for, never killed: killing
+    it would make the other side record a failed fence and refuse its deploy.
+    Only a leftover, one idle past HELPER_GRACE or still there after the helper
+    timeout, is removed.
     """
     from docker.errors import NotFound
 
+    deadline = time.monotonic() + HELPER_TIMEOUT + HELPER_GRACE
+    while True:
+        try:
+            found = client.containers.get(HELPER_NAME)
+            if (getattr(found, "labels", None) or {}).get(PROBE_LABEL) != "fence":
+                return
+            if not _helper_busy(found.attrs or {}, time.time()) or time.monotonic() > deadline:
+                found.remove(force=True, v=True)
+                return
+        except NotFound:
+            return
+        except Exception as exc:  # noqa: BLE001 - logged; the run that follows fails on the name
+            logger.warning("Could not remove %s: %s", HELPER_NAME, exc)
+            return
+        time.sleep(HELPER_POLL)
+
+
+def _remove_own(container: Any) -> None:
+    """Remove the helper this process started, by its own ID, never by name."""
+    from docker.errors import NotFound
+
     try:
-        found = client.containers.get(HELPER_NAME)
-        if (getattr(found, "labels", None) or {}).get(PROBE_LABEL) == "fence":
-            found.remove(force=True, v=True)
+        container.remove(force=True, v=True)
     except NotFound:
         pass
-    except Exception as exc:  # noqa: BLE001 - logged; the next run removes it by name first
+    except Exception as exc:  # noqa: BLE001 - logged; the next run clears it as a leftover
         logger.warning("Could not remove %s: %s", HELPER_NAME, exc)
 
 
 def run_helper(client: Any, action: str, subnet: str, filtered: bool = True) -> dict[str, Any]:
     """Run the helper to completion and return what it printed. Removed on every path."""
+    from docker.errors import APIError
+
     with _lock:
-        _remove_helper(client)
+        container = None
         try:
-            container = client.containers.run(
-                image=_helper_image(client),
-                name=HELPER_NAME,
-                entrypoint=["python", "-c", HELPER_SCRIPT],
-                command=[action],
-                environment={
-                    "FENCE_IFACE": BRIDGE_INTERFACE,
-                    "FENCE_SUBNET": subnet,
-                    "FENCE_FILTER": "1" if filtered else "0",
-                    "FENCE_BACKEND": _BACKEND,
-                    "FENCE_EGRESS": _egress["ip"] or "",
-                },
-                network_mode="host",
-                user="0:0",
-                read_only=True,
-                # iptables-legacy takes its lock in /run.
-                tmpfs={"/run": "size=1m"},
-                cap_drop=["ALL"],
-                # NET_RAW only matters to iptables-legacy, which talks to the
-                # kernel over a raw socket. Both run our own script, never an earner.
-                cap_add=["NET_ADMIN", "NET_RAW"],
-                security_opt=["no-new-privileges:true"],
-                privileged=False,
-                pids_limit=64,
-                mem_limit="64m",
-                healthcheck={"test": ["NONE"]},
-                labels={PROBE_LABEL: "fence"},
-                detach=True,
-            )
+            # Twice: another process can create its helper between our clear and our run.
+            for attempt in range(2):
+                _clear_helper(client)
+                try:
+                    container = _create_helper(client, action, subnet, filtered)
+                    break
+                except APIError as exc:
+                    if attempt or getattr(exc, "status_code", None) != 409:
+                        raise
+            # Created and started apart, so a start that fails still leaves an
+            # ID to remove: nothing is ever removed by name after the create.
+            container.start()
             status = container.wait(timeout=HELPER_TIMEOUT)
             text = container.logs(stdout=True, stderr=False).decode(errors="replace").strip()
             if (status or {}).get("StatusCode") != 0:
@@ -831,7 +879,51 @@ def run_helper(client: Any, action: str, subnet: str, filtered: bool = True) ->
                 )
             return json.loads(text.splitlines()[-1])
         finally:
-            _remove_helper(client)
+            if container is not None:
+                _remove_own(container)
+
+
+def _create_helper(client: Any, action: str, subnet: str, filtered: bool) -> Any:
+    """Create the helper container, not started, pulling its image once if it is missing."""
+    from docker.errors import ImageNotFound
+
+    image = _helper_image(client)
+    try:
+        return client.containers.create(image=image, **_helper_spec(action, subnet, filtered))
+    except ImageNotFound:
+        client.images.pull(image)
+        return client.containers.create(image=image, **_helper_spec(action, subnet, filtered))
+
+
+def _helper_spec(action: str, subnet: str, filtered: bool) -> dict[str, Any]:
+    """The helper's container settings: host networking, NET_ADMIN and NET_RAW, every other capability dropped."""
+    return dict(
+        name=HELPER_NAME,
+        entrypoint=["python", "-c", HELPER_SCRIPT],
+        command=[action],
+        environment={
+            "FENCE_IFACE": BRIDGE_INTERFACE,
+            "FENCE_SUBNET": subnet,
+            "FENCE_FILTER": "1" if filtered else "0",
+            "FENCE_BACKEND": _BACKEND,
+            "FENCE_EGRESS": _egress["ip"] or "",
+        },
+        network_mode="host",
+        user="0:0",
+        read_only=True,
+        # iptables-legacy takes its lock in /run.
+        tmpfs={"/run": "size=1m"},
+        cap_drop=["ALL"],
+        # NET_RAW only matters to iptables-legacy, which talks to the
+        # kernel over a raw socket. Both run our own script, never an earner.
+        cap_add=["NET_ADMIN", "NET_RAW"],
+        security_opt=["no-new-privileges:true"],
+        privileged=False,
+        pids_limit=64,
+        mem_limit="64m",
+        healthcheck={"test": ["NONE"]},
+        labels={PROBE_LABEL: "fence"},
+    )
 
 
 def _set_fence(state: str, result: dict[str, Any] | None = None, error: str = "") -> None:

tests/test_lan_fence.py

@@ -18,6 +18,7 @@
 import subprocess
 import sys
 from contextlib import asynccontextmanager
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -26,7 +27,7 @@
 os.environ.setdefault("CASHPILOT_API_KEY", "test-fleet-key")
 
 import pytest
-from docker.errors import NotFound
+from docker.errors import APIError, ImageNotFound, NotFound
 from fastapi.testclient import TestClient
 
 from app import isolation_check, lan_fence, orchestrator, worker_api
@@ -860,13 +861,13 @@ def test_the_next_deploy_lands_behind_the_fence(self):
 def test_the_helper_gets_the_egress_ip():
     lan_fence.set_egress_ip("81.61.1.9")
     client = MagicMock()
-    helper = client.containers.run.return_value
+    helper = client.containers.create.return_value
     helper.wait.return_value = {"StatusCode": 0}
     helper.logs.return_value = b'{"ok": true}'
     client.containers.get.side_effect = NotFound("gone")
     with patch.object(lan_fence, "_helper_image", return_value="img"):
         lan_fence.run_helper(client, "apply", "172.18.0.0/16")
-    assert client.containers.run.call_args.kwargs["environment"]["FENCE_EGRESS"] == "81.61.1.9"
+    assert client.containers.create.call_args.kwargs["environment"]["FENCE_EGRESS"] == "81.61.1.9"
     lan_fence.set_egress_ip("10.0.0.1")
     assert lan_fence._egress["ip"] == "81.61.1.9", "a LAN answer is a failed lookup: never dropped, the last IP kept"
 
@@ -929,14 +930,117 @@ def test_a_foreign_container_by_the_helpers_name_is_left_alone(self):
         foreign = MagicMock(labels={})
         client = MagicMock()
         client.containers.get.return_value = foreign
-        lan_fence._remove_helper(client)
+        lan_fence._clear_helper(client)
         foreign.remove.assert_not_called()
-        ours = MagicMock(labels={lan_fence.PROBE_LABEL: "fence"})
+        ours = _helper_container("exited", finished=_ago(3600))
         client.containers.get.return_value = ours
-        lan_fence._remove_helper(client)
+        lan_fence._clear_helper(client)
         ours.remove.assert_called_once_with(force=True, v=True)
 
 
+def _ago(seconds):
+    return (datetime.now(UTC) - timedelta(seconds=seconds)).isoformat().replace("+00:00", "123Z")
+
+
+def _helper_container(status, created=None, finished="0001-01-01T00:00:00Z"):
+    c = MagicMock(labels={lan_fence.PROBE_LABEL: "fence"})
+    c.attrs = {
+        "Created": created or _ago(3600),
+        "State": {"Status": status, "Running": status == "running", "Restarting": False, "FinishedAt": finished},
+    }
+    return c
+
+
+def _run_helper(client):
+    mine = client.containers.create.return_value
+    mine.wait.return_value = {"StatusCode": 0}
+    mine.logs.return_value = b'{"ok": true}'
+    with patch.object(lan_fence, "_helper_image", return_value="img"), patch.object(lan_fence, "HELPER_POLL", 0):
+        return lan_fence.run_helper(client, "apply", "172.18.0.0/16"), mine
+
+
+class TestAnotherProcesssHelper:
+    """The Desktop app runs the same helper, by the same name and label, on a host that can run both."""
+
+    @pytest.mark.parametrize(
+        "theirs",
+        [
+            pytest.param(lambda: _helper_container("running"), id="running"),
+            pytest.param(lambda: _helper_container("created", created=_ago(1)), id="just-created"),
+            pytest.param(lambda: _helper_container("exited", finished=_ago(1)), id="just-exited"),
+        ],
+    )
+    def test_a_helper_still_in_use_is_waited_for_never_killed(self, theirs):
+        other = theirs()
+        client = MagicMock()
+        client.containers.get.side_effect = [other, other, NotFound("gone")]
+        result, mine = _run_helper(client)
+        assert result == {"ok": True}
+        other.remove.assert_not_called()
+        assert client.containers.get.call_count == 3, "it looked again until the other side removed it"
+        client.containers.create.assert_called_once()
+        mine.remove.assert_called_once_with(force=True, v=True)
+
+    def test_a_leftover_is_removed(self):
+        stale = _helper_container("exited", finished=_ago(3600))
+        client = MagicMock()
+        client.containers.get.return_value = stale
+        _run_helper(client)
+        stale.remove.assert_called_once_with(force=True, v=True)
+
+    def test_one_still_there_after_the_helper_timeout_is_removed(self):
+        stuck = _helper_container("running")
+        client = MagicMock()
+        client.containers.get.return_value = stuck
+        with patch.object(lan_fence, "HELPER_TIMEOUT", 0), patch.object(lan_fence, "HELPER_GRACE", 0):
+            _run_helper(client)
+        stuck.remove.assert_called_once_with(force=True, v=True)
+
+    def test_only_its_own_helper_is_removed_afterwards(self):
+        """By its own object: a name lookup after the run could find the other side's new helper."""
+        client = MagicMock()
+        client.containers.get.side_effect = NotFound("gone")
+        _, mine = _run_helper(client)
+        mine.remove.assert_called_once_with(force=True, v=True)
+        assert client.containers.get.call_count == 1, "no lookup by name after the run"
+
+    def test_a_name_conflict_from_a_racing_create_is_tried_once_more(self):
+        client = MagicMock()
+        client.containers.get.side_effect = NotFound("gone")
+        mine = client.containers.create.return_value
+        response = MagicMock(status_code=409)
+        client.containers.create.side_effect = [APIError("Conflict", response=response), mine]
+        result, _ = _run_helper(client)
+        assert result == {"ok": True}
+        assert client.containers.create.call_count == 2
+
+    def test_a_helper_that_never_started_leaves_nothing_to_remove(self):
+        client = MagicMock()
+        client.containers.get.side_effect = NotFound("gone")
+        client.containers.create.side_effect = APIError("boom", response=MagicMock(status_code=500))
+        with pytest.raises(APIError):
+            _run_helper(client)
+        assert client.containers.create.call_count == 1
+
+    def test_a_helper_that_fails_to_start_is_removed_by_its_id(self):
+        client = MagicMock()
+        client.containers.get.side_effect = NotFound("gone")
+        mine = client.containers.create.return_value
+        mine.start.side_effect = APIError("no such image layer", response=MagicMock(status_code=500))
+        with pytest.raises(APIError):
+            _run_helper(client)
+        mine.remove.assert_called_once_with(force=True, v=True)
+
+    def test_a_missing_image_is_pulled_once(self):
+        client = MagicMock()
+        client.containers.get.side_effect = NotFound("gone")
+        mine = client.containers.create.return_value
+        client.containers.create.side_effect = [ImageNotFound("img"), mine]
+        result, _ = _run_helper(client)
+        assert result == {"ok": True}
+        client.images.pull.assert_called_once_with("img")
+
+
 class TestEnable:
     def test_off_setting_is_refused(self):
         with (