From b9458e8360b0b063656013d24772d79151d7824d Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Sat, 7 Nov 2026 14:35:31 +0100 Subject: [PATCH 1/2] fix(proxies): keep the running copy when a redeploy's tunnel cannot start deploy() removed the old earner and tunnel before the new tunnel started, so a tunnel that failed to start (an image missing locally, a proxy that does not answer) left the proxy with no copy at all. The new tunnel now starts under a staged name next to the running copy. Only once it has proved an exit does the old pair go and the tunnel take the final name, under a lock the heartbeat's status() shares. The heartbeat never reports a staged tunnel, and a worker start sweeps one a dead deploy left behind. --- app/proxy_tunnel.py | 68 ++++++++++++++++++++++++---- tests/test_proxy_tunnel.py | 90 ++++++++++++++++++++++++++++++++++++-- 2 files changed, 145 insertions(+), 13 deletions(-) diff --git a/app/proxy_tunnel.py b/app/proxy_tunnel.py index 23a6331b..54930c71 100644 --- a/app/proxy_tunnel.py +++ b/app/proxy_tunnel.py @@ -31,6 +31,7 @@ import re import socket import tarfile +import threading import time import uuid from typing import Any @@ -61,6 +62,13 @@ READY_TIMEOUT = 45.0 SECRET_DIR = "/run/cashpilot-secret" _FLOATING_TAGS = {"", "latest", "stable", "edge", "main", "master"} +#: A redeploy starts the new sidecar under its tunnel name plus this suffix, next +#: to the copy that runs now, and gives it the final name only once the old pair +#: is gone. The heartbeat (status()) never reports a staged sidecar. +STAGED_SUFFIX = "-next" +# deploy() swapping the old pair for the staged one, and status(), take turns: a +# heartbeat must not start an earner that is being removed, or read a half swap. +_swap_lock = threading.Lock() class ProxyTunnelError(RuntimeError): @@ -357,7 +365,13 @@ def _get(client: Any, name: str) -> Any | None: def start_sidecar( - client: Any, name: str, slug: str, proxy_id: int | None, proxy: dict[str, Any], own_ip: str | None + client: Any, + name: str, + slug: str, + proxy_id: int | None, + proxy: dict[str, Any], + own_ip: str | None, + hostname: str | None = None, ) -> tuple[Any, dict[str, Any]]: """Create, provision and start a sidecar, then wait for its exit. Removed on any failure.""" network = orchestrator.container_network("bridge") @@ -367,7 +381,7 @@ def start_sidecar( client.images.pull(image) except APIError as exc: logger.warning("Could not pull %s: %s (trying the local copy)", image, exc) - kwargs = sidecar_kwargs(name, slug, proxy_id, network, hostname=name.removeprefix(CONTAINER_PREFIX)[:63]) + kwargs = sidecar_kwargs(name, slug, proxy_id, network, hostname=hostname or _hostname(name)) kwargs.pop("detach") container = client.containers.create(**kwargs) try: @@ -384,19 +398,38 @@ def start_sidecar( return container, exit_info +def _hostname(name: str) -> str: + return name.removeprefix(CONTAINER_PREFIX)[:63] + + def deploy(slug: str, proxy_id: int, earner: dict[str, Any], proxy: dict[str, Any], own_ip: str | None) -> dict: """Deploy (or replace) the proxied copy of ``slug`` through proxy ``proxy_id``. - Order: the old pair goes (earner first), the sidecar starts and proves an - exit, and only then is the earner created inside its namespace. + Order: the new sidecar starts under a staged name and proves an exit while + the old pair keeps running; then the old pair goes (earner first), the + sidecar takes the final name, and only then is the earner created inside + its namespace. A sidecar that cannot start therefore costs nothing. """ client = orchestrator._get_client() - # Checked before the old pair goes: settings that cannot work (a host that - # does not resolve, no tunnel image) must not cost the copy running now. + # Checked first: settings that cannot work (a host that does not resolve, + # no tunnel image) fail fast, before a sidecar is created. secret_files(proxy, own_ip, f"{slug}-p{proxy_id}") tunnel_image() - remove(slug, proxy_id) - sidecar, exit_info = start_sidecar(client, tunnel_name(slug, proxy_id), slug, proxy_id, proxy, own_ip) + final = tunnel_name(slug, proxy_id) + staged = final + STAGED_SUFFIX + # Left by a deploy that died midway, on this name only if it is ours. + if (leftover := _get(client, staged)) is not None: + if not _is_ours(leftover, slug, proxy_id): + raise ProxyTunnelError(f"A container named {staged} that CashPilot did not create is in the way") + _remove_quietly(leftover, 2) + sidecar, exit_info = start_sidecar(client, staged, slug, proxy_id, proxy, own_ip, hostname=_hostname(final)) + try: + with _swap_lock: + remove(slug, proxy_id) + sidecar.rename(final) + except Exception: + _remove_quietly(sidecar, 2) + raise try: container_id = orchestrator.deploy_raw( slug=slug, @@ -529,7 +562,8 @@ def remove_all(slug: str) -> list[str]: def sweep_test_tunnels() -> list[str]: - """Remove Test tunnels a worker that stopped mid-test left behind. + """Remove Test tunnels a worker that stopped mid-test left behind, and staged + sidecars of a deploy it stopped in. Run at worker start, when no test of this worker can be in flight. """ @@ -544,6 +578,15 @@ def sweep_test_tunnels() -> list[str]: continue _remove_quietly(container, 2) swept.append(container.name) + # And staged sidecars of a deploy the worker died in: they hold the proxy + # login, and the heartbeat never reports them. + staged = client.containers.list( + all=True, filters={"label": [f"{LABEL_MANAGED}=true", f"{LABEL_ROLE}={ROLE_TUNNEL}", LABEL_PROXY]} + ) + for container in staged: + if str(container.name).endswith(STAGED_SUFFIX): + _remove_quietly(container, 2) + swept.append(container.name) return swept @@ -600,6 +643,11 @@ def status(collect_stats: bool = True) -> list[dict[str, Any]]: stats carry no network counters, and every byte it moves crosses the sidecar's eth0. """ + with _swap_lock: + return _status(collect_stats) + + +def _status(collect_stats: bool) -> list[dict[str, Any]]: try: client = orchestrator._get_client() found = client.containers.list(all=True, filters={"label": [f"{LABEL_MANAGED}=true", LABEL_PROXY]}) @@ -608,6 +656,8 @@ def status(collect_stats: bool = True) -> list[dict[str, Any]]: return [] pairs: dict[tuple[str, str], dict[str, Any]] = {} for container in found: + if str(getattr(container, "name", "")).endswith(STAGED_SUFFIX): + continue # a redeploy's sidecar, not promoted yet labels = container.labels or {} key = (labels.get(LABEL_SERVICE, ""), labels.get(LABEL_PROXY, "")) role = "tunnel" if labels.get(LABEL_ROLE) == ROLE_TUNNEL else "earner" diff --git a/tests/test_proxy_tunnel.py b/tests/test_proxy_tunnel.py index 7fea50a9..6fa558b7 100644 --- a/tests/test_proxy_tunnel.py +++ b/tests/test_proxy_tunnel.py @@ -298,21 +298,94 @@ def fake_deploy_raw(**kw): assert "ports" not in kw and "volumes" not in kw return "earner-id" + sidecar.rename.side_effect = lambda name: calls.append(f"renamed {name}") + client = MagicMock() + client.containers.get.side_effect = NotFound("gone") with ( - patch.object(orchestrator, "_get_client", return_value=MagicMock()), + patch.object(orchestrator, "_get_client", return_value=client), patch.object(proxy_tunnel, "tunnel_image", return_value="drumsergio/cashpilot-tunnel:1.64.0"), patch.object(proxy_tunnel, "remove", side_effect=lambda *a: calls.append("old-removed") or []), - patch.object(proxy_tunnel, "start_sidecar", side_effect=fake_start_sidecar), + patch.object(proxy_tunnel, "start_sidecar", side_effect=fake_start_sidecar) as start, patch.object(orchestrator, "deploy_raw", side_effect=fake_deploy_raw), ): result = proxy_tunnel.deploy("honeygain", 3, {"image": "honeygain/honeygain:1"}, _socks(), None) - assert calls == ["old-removed", "sidecar-ready", "earner"] + # The new sidecar proves its exit while the old pair still runs, under a staged name. + assert calls == ["sidecar-ready", "old-removed", "renamed cashpilot-tunnel-honeygain-p3", "earner"] + assert start.call_args.args[1] == "cashpilot-tunnel-honeygain-p3-next" + assert start.call_args.kwargs["hostname"] == "tunnel-honeygain-p3" assert result["container_id"] == "earner-id" + def test_a_sidecar_that_cannot_start_keeps_the_running_copy(self): + client = MagicMock() + client.containers.get.side_effect = NotFound("gone") + with ( + patch.object(orchestrator, "_get_client", return_value=client), + patch.object(proxy_tunnel, "tunnel_image", return_value="drumsergio/cashpilot-tunnel:1.64.0"), + patch.object(proxy_tunnel, "remove") as remove, + patch.object(proxy_tunnel, "start_sidecar", side_effect=proxy_tunnel.ProxyTunnelError("No such image")), + pytest.raises(proxy_tunnel.ProxyTunnelError), + ): + proxy_tunnel.deploy("honeygain", 3, {"image": "honeygain/honeygain:1"}, _socks(), None) + remove.assert_not_called() + + def test_a_failed_promotion_removes_the_staged_sidecar(self): + sidecar = MagicMock(id="sidecar-id") + sidecar.rename.side_effect = RuntimeError("409 Conflict") + client = MagicMock() + client.containers.get.side_effect = NotFound("gone") + with ( + patch.object(orchestrator, "_get_client", return_value=client), + patch.object(proxy_tunnel, "tunnel_image", return_value="drumsergio/cashpilot-tunnel:1.64.0"), + patch.object(proxy_tunnel, "remove", return_value=[]), + patch.object(proxy_tunnel, "start_sidecar", return_value=(sidecar, {"ip": "93.184.215.14"})), + patch.object(orchestrator, "deploy_raw") as deploy_raw, + pytest.raises(RuntimeError), + ): + proxy_tunnel.deploy("honeygain", 3, {"image": "honeygain/honeygain:1"}, _socks(), None) + sidecar.remove.assert_called_once_with(force=True, v=True) + deploy_raw.assert_not_called() + + def test_a_staged_sidecar_left_behind_is_replaced_but_never_a_stranger(self): + leftover = _container("cashpilot-tunnel-honeygain-p3-next", "honeygain", proxy=3, role="tunnel") + stranger = _container("cashpilot-tunnel-honeygain-p3-next", "other") + sidecar = MagicMock(id="sidecar-id") + for found, ok in ((leftover, True), (stranger, False)): + client = MagicMock() + client.containers.get.return_value = found + with ( + patch.object(orchestrator, "_get_client", return_value=client), + patch.object(proxy_tunnel, "tunnel_image", return_value="drumsergio/cashpilot-tunnel:1.64.0"), + patch.object(proxy_tunnel, "remove", return_value=[]), + patch.object(proxy_tunnel, "start_sidecar", return_value=(sidecar, {"ip": "93.184.215.14"})), + patch.object(orchestrator, "deploy_raw", return_value="earner-id"), + ): + if ok: + proxy_tunnel.deploy("honeygain", 3, {"image": "honeygain/honeygain:1"}, _socks(), None) + else: + with pytest.raises(proxy_tunnel.ProxyTunnelError, match="did not create"): + proxy_tunnel.deploy("honeygain", 3, {"image": "honeygain/honeygain:1"}, _socks(), None) + assert found.remove.called is ok + + def test_the_heartbeat_never_reports_a_staged_sidecar(self): + earner = _container("cashpilot-x-p1", "x", proxy=1) + tunnel = _container("cashpilot-tunnel-x-p1", "x", proxy=1, role="tunnel") + staged = _container("cashpilot-tunnel-x-p1-next", "x", proxy=1, role="tunnel") + staged.id = "id-staged" + client = MagicMock() + client.containers.list.return_value = [earner, tunnel, staged] + with ( + patch.object(orchestrator, "_get_client", return_value=client), + patch.object(proxy_tunnel, "_read_exit", return_value=None), + ): + entries = proxy_tunnel.status(collect_stats=False) + assert [(e["slug"], e["tunnel_container_id"]) for e in entries] == [("x", tunnel.short_id)] + def test_earner_failure_removes_the_sidecar(self): sidecar = MagicMock(id="sidecar-id") + client = MagicMock() + client.containers.get.side_effect = NotFound("gone") with ( - patch.object(orchestrator, "_get_client", return_value=MagicMock()), + patch.object(orchestrator, "_get_client", return_value=client), patch.object(proxy_tunnel, "tunnel_image", return_value="drumsergio/cashpilot-tunnel:1.64.0"), patch.object(proxy_tunnel, "remove", return_value=[]), patch.object(proxy_tunnel, "start_sidecar", return_value=(sidecar, {"ip": "93.184.215.14"})), @@ -882,6 +955,15 @@ def test_leftover_test_tunnels_are_swept_and_copies_are_not(self): leftover.remove.assert_called_once_with(force=True, v=True) copy_tunnel.remove.assert_not_called() + def test_a_staged_sidecar_of_a_deploy_that_died_is_swept(self): + staged = _container("cashpilot-tunnel-x-p1-next", "x", proxy=1, role="tunnel") + live = _container("cashpilot-tunnel-x-p1", "x", proxy=1, role="tunnel") + client = MagicMock() + client.containers.list.side_effect = [[], [staged, live]] + with patch.object(orchestrator, "_get_client", return_value=client): + assert proxy_tunnel.sweep_test_tunnels() == ["cashpilot-tunnel-x-p1-next"] + live.remove.assert_not_called() + def test_a_probe_answer_that_is_not_an_ip_is_not_ready(self): sidecar = MagicMock(status="running") sidecar.exec_run.return_value = (0, b'{"ip":"e","country":""}\n') From 27b24563995181d1c5a919879afb0987a0283481 Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Sat, 7 Nov 2026 15:11:31 +0100 Subject: [PATCH 2/2] fix(proxies): sweep only stale staged sidecars, and run one deploy per copy at a time --- app/proxy_tunnel.py | 35 ++++++++++++++++++++++++++++++++--- tests/test_proxy_tunnel.py | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 65 insertions(+), 3 deletions(-) diff --git a/app/proxy_tunnel.py b/app/proxy_tunnel.py index 54930c71..60eadea5 100644 --- a/app/proxy_tunnel.py +++ b/app/proxy_tunnel.py @@ -34,6 +34,7 @@ import threading import time import uuid +from datetime import UTC, datetime from typing import Any from urllib.parse import quote @@ -69,6 +70,28 @@ # deploy() swapping the old pair for the staged one, and status(), take turns: a # heartbeat must not start an earner that is being removed, or read a half swap. _swap_lock = threading.Lock() +# One deploy per copy at a time: a second one would take the first one's staged +# sidecar for a leftover and remove it. +_deploy_locks: dict[tuple[str, int], threading.Lock] = {} +_deploy_locks_guard = threading.Lock() +#: A staged sidecar older than this is a leftover: a deploy is past its ready +#: timeout by then. Younger ones may belong to a deploy still running. +STAGED_STALE_SECONDS = 300 + + +def _deploy_lock(slug: str, proxy_id: int) -> threading.Lock: + with _deploy_locks_guard: + return _deploy_locks.setdefault((slug, proxy_id), threading.Lock()) + + +def _age_seconds(container: Any) -> float: + """Seconds since the container was created; 0 when Docker does not say.""" + raw = str((getattr(container, "attrs", None) or {}).get("Created") or "") + try: + created = datetime.fromisoformat(re.sub(r"(\.\d{6})\d+", r"\1", raw).replace("Z", "+00:00")) + except ValueError: + return 0.0 + return (datetime.now(UTC) - created).total_seconds() class ProxyTunnelError(RuntimeError): @@ -410,6 +433,11 @@ def deploy(slug: str, proxy_id: int, earner: dict[str, Any], proxy: dict[str, An sidecar takes the final name, and only then is the earner created inside its namespace. A sidecar that cannot start therefore costs nothing. """ + with _deploy_lock(slug, proxy_id): + return _deploy(slug, proxy_id, earner, proxy, own_ip) + + +def _deploy(slug: str, proxy_id: int, earner: dict[str, Any], proxy: dict[str, Any], own_ip: str | None) -> dict: client = orchestrator._get_client() # Checked first: settings that cannot work (a host that does not resolve, # no tunnel image) fail fast, before a sidecar is created. @@ -578,13 +606,14 @@ def sweep_test_tunnels() -> list[str]: continue _remove_quietly(container, 2) swept.append(container.name) - # And staged sidecars of a deploy the worker died in: they hold the proxy - # login, and the heartbeat never reports them. + # And staged sidecars of a deploy that died: they hold the proxy login, and + # the heartbeat never reports them. Only stale ones, so a deploy another + # worker on this daemon is running keeps its sidecar. staged = client.containers.list( all=True, filters={"label": [f"{LABEL_MANAGED}=true", f"{LABEL_ROLE}={ROLE_TUNNEL}", LABEL_PROXY]} ) for container in staged: - if str(container.name).endswith(STAGED_SUFFIX): + if str(container.name).endswith(STAGED_SUFFIX) and _age_seconds(container) >= STAGED_STALE_SECONDS: _remove_quietly(container, 2) swept.append(container.name) return swept diff --git a/tests/test_proxy_tunnel.py b/tests/test_proxy_tunnel.py index 6fa558b7..e76907dd 100644 --- a/tests/test_proxy_tunnel.py +++ b/tests/test_proxy_tunnel.py @@ -964,6 +964,39 @@ def test_a_staged_sidecar_of_a_deploy_that_died_is_swept(self): assert proxy_tunnel.sweep_test_tunnels() == ["cashpilot-tunnel-x-p1-next"] live.remove.assert_not_called() + def test_a_young_staged_sidecar_is_left_to_its_deploy(self): + from datetime import UTC, datetime + + young = _container("cashpilot-tunnel-x-p1-next", "x", proxy=1, role="tunnel") + young.attrs["Created"] = datetime.now(UTC).isoformat().replace("+00:00", "123Z") + client = MagicMock() + client.containers.list.side_effect = [[], [young]] + with patch.object(orchestrator, "_get_client", return_value=client): + assert proxy_tunnel.sweep_test_tunnels() == [] + young.remove.assert_not_called() + + def test_two_deploys_of_one_copy_take_turns(self): + import threading + + inside, overlap = [], [] + gate = threading.Event() + + def slow(*_a, **_kw): + if inside: + overlap.append(True) + inside.append(True) + gate.wait(0.2) + inside.pop() + return {} + + with patch.object(proxy_tunnel, "_deploy", side_effect=slow): + threads = [threading.Thread(target=proxy_tunnel.deploy, args=("x", 1, {}, {}, None)) for _ in range(2)] + for t in threads: + t.start() + for t in threads: + t.join() + assert not overlap + def test_a_probe_answer_that_is_not_an_ip_is_not_ready(self): sidecar = MagicMock(status="running") sidecar.exec_run.return_value = (0, b'{"ip":"e","country":""}\n')