fix(worker): stop tearing down a node Docker is still starting

app/catalog.py

@@ -81,8 +81,12 @@ def parse_port(mapping: object) -> tuple[str, int | list[str | int]]:
 NETNS_CAPABILITIES = frozenset({"NET_ADMIN", "SYS_ADMIN"})
 
 #: The most UDP ports one udp_port_range may publish. Each published port is a
-#: rule and, with Docker's userland proxy, a process on the host.
-MAX_UDP_PORTS = 1000
+#: rule and, with Docker's userland proxy, a process on the host, and each one
+#: slows the container's start. Measured on a live node (Nov 2026): 300 ports
+#: start in about 12 s, about 350 take 30 s, where the socket proxy stops
+#: waiting for Docker's answer, and 1,000 take about 51 s. 300 keeps a deploy's
+#: start well inside the proxy's 30 s on a slower host too.
+MAX_UDP_PORTS = 300
 
 
 def is_port_range(value: str) -> bool:
@@ -114,7 +118,12 @@ def expand_udp_range(value: str, cap: int = MAX_UDP_PORTS) -> dict[str, int]:
     start, _, end = value.partition(":")
     first, last = int(start), int(end)
     if last - first + 1 > cap:
-        raise ValueError(f"{value!r} spans {last - first + 1} ports; at most {cap} can be published")
+        raise ValueError(
+            f"{value!r} spans {last - first + 1} ports; at most {cap} can be published, since every published "
+            f"port slows the container's start and a wider range runs past what a deploy can wait for. Choose a "
+            f"range of {cap} ports or fewer, such as {first}:{min(first + cap - 1, 65535)}, and forward the same "
+            "range on your router"
+        )
     return {f"{port}/udp": port for port in range(first, last + 1)}
 
 

app/constants.py

@@ -58,6 +58,13 @@
 #: then fails the whole start.
 EPHEMERAL_PORTS = (32768, 60999)
 
+#: How long a deploy waits, counted from its create call, for Docker to finish
+#: starting a container when the call itself got no answer. The socket proxy
+#: gives up on a call after 30 s and docker-py after 60 s, while Docker carries
+#: on: a node publishing 1,000 UDP ports took about 51 s to start. The UI adds
+#: it to its own wait for the worker's deploy answer.
+START_SETTLE_SECONDS = 120
+
 
 def drift_warnings_enabled() -> bool:
     """CASHPILOT_DRIFT_WARNINGS: off stops the worker reporting container settings

app/drift.py

@@ -166,13 +166,22 @@ def move_advice(
     key = str(rng.get("env") or "") if isinstance(rng, dict) else ""
     value = str(((recorded or {}).get("env") or {}).get(key) or "").strip() if key else ""
     m = re.fullmatch(r"(\d{1,5}):(\d{1,5})", value)
-    if m and int(m[1]) <= EPHEMERAL_PORTS[1] and int(m[2]) >= EPHEMERAL_PORTS[0]:
-        low = EPHEMERAL_PORTS[0]
-        default = next((str(v.get("default") or "") for v in docker_conf.get("env") or [] if v.get("key") == key), "")
-        example = ""
-        d = re.fullmatch(r"(\d{1,5}):(\d{1,5})", default)
-        if d and int(d[2]) + 100 < low:
-            example = f", such as {int(d[1]) + 100}:{int(d[2]) + 100}"
+    cap = int(rng.get("max_ports") or catalog.MAX_UDP_PORTS) if isinstance(rng, dict) else catalog.MAX_UDP_PORTS
+    low = EPHEMERAL_PORTS[0]
+    default = next((str(v.get("default") or "") for v in docker_conf.get("env") or [] if v.get("key") == key), "")
+    example = ""
+    d = re.fullmatch(r"(\d{1,5}):(\d{1,5})", default)
+    if d and int(d[2]) + 100 < low:
+        example = f", such as {int(d[1]) + 100}:{int(d[2]) + 100}"
+    if m and int(m[2]) - int(m[1]) + 1 > cap:
+        # Wider than a deploy publishes: the redeploy is refused, so reserving it is no way out.
+        advice["warning"] = (
+            f"Its recorded {key}, {value}, spans {int(m[2]) - int(m[1]) + 1} ports, more than the {cap} a deploy "
+            f"publishes, so the redeploy is refused until it is narrower. Before you redeploy, set {key} to a free "
+            f"range of at most {cap} ports below {low} other than the default{example} (a value equal to the default "
+            "keeps the recorded one), and change the router forward to match."
+        )
+    elif m and int(m[1]) <= EPHEMERAL_PORTS[1] and int(m[2]) >= EPHEMERAL_PORTS[0]:
         advice["warning"] = (
             f"Its recorded {key}, {value}, overlaps Linux's ephemeral range ({low}-{EPHEMERAL_PORTS[1]}), which the "
             "redeploy keeps: on its own network every port in it is published, and one a host program holds fails "
@@ -267,7 +276,9 @@ def container_drift(
                 (v.get("default") for v in docker_conf.get("env") or [] if v.get("key") == rng.get("env")), ""
             )
             with contextlib.suppress(ValueError):
-                want_ports |= set(catalog.expand_udp_range(str(value or ""), int(rng.get("max_ports") or 1000)))
+                want_ports |= set(
+                    catalog.expand_udp_range(str(value or ""), int(rng.get("max_ports") or catalog.MAX_UDP_PORTS))
+                )
         if missing := want_ports - set(shape.get("ports") or []):
             found.append(f"does not publish port {_names(missing)}")
     if gated and shape.get("restart_policy") not in (None, "no"):

app/lan_fence.py

@@ -1268,8 +1268,15 @@ def guard() -> dict[str, str]:
     return held
 
 
-#: Why the guard holds a container a failed deploy left behind.
-FAILED_DEPLOY = "its last deploy failed and left this container behind. Redeploy it from the dashboard"
+#: Why the guard holds a container under a name whose last deploy failed. The
+#: record is by name, so it also holds the old container put back by hand, or
+#: one started with docker start: the worker cannot tell it from the failed
+#: one, and the operator has to hear that only a deploy from CashPilot lifts it.
+FAILED_DEPLOY = (
+    "its last deploy from CashPilot failed, so the worker keeps any container under this name stopped, one put "
+    "back or started by hand included, until a deploy from CashPilot succeeds. Redeploy it from the CashPilot "
+    "dashboard; starting it with Docker does not lift this"
+)
 
 
 #: A start the guard could not read back: stopped, but tried again next pass.

app/main.py

@@ -283,6 +283,9 @@ async def _get_all_worker_containers(workers: list[dict[str, Any]] | None = None
                             # When that refused run ended, to tell a refusal
                             # older than the last balance read (stale) apart.
                             "fatal_at": c.get("fatal_at"),
+                            # Why the worker's fence guard keeps this copy stopped, and
+                            # what brings it back (lan_fence.guard). Absent when it does not.
+                            "held": c.get("held"),
                             # What the container was created with, for the drift
                             # check (app/drift.py). None from a worker too old to send it.
                             "shape": c.get("shape"),
@@ -586,6 +589,10 @@ def _container_down_message(slug: str, copies: list[dict[str, Any]]) -> str:
     how = f" (exit code {', '.join(str(code) for code in codes)})" if codes else ""
     message = f"{name} has not been running since {since} UTC on {nodes}{how}, so it is not earning."
     if catalog.fence_gated(catalog.get_service(slug)):
+        # What the worker's guard says holds it; a reason only a redeploy clears
+        # must reach the operator here, not only on the Fleet page.
+        if held := sorted({str(c["held"]) for c in copies if c.get("held")}):
+            return message + f" The worker keeps it stopped: {'; '.join(held)}."
         # No restart policy on purpose: the worker starts it once the LAN fence is confirmed.
         return message + (
             " The worker starts it as soon as the LAN fence is confirmed; while it stays down, the Fleet page "
@@ -2820,6 +2827,7 @@ async def api_services_deployed(request: Request) -> list[dict[str, Any]]:
                 # What a windowed health signal says is wrong with this copy.
                 "hints": _windowed_hints.get((slug, inst.get("_worker_id"), int(inst.get("instance") or 1)), []),
                 "fatal": inst.get("fatal"),
+                "held": inst.get("held"),
             }
             # Only containers CashPilot made (constants.CASHPILOT_DEPLOYED): a
             # container the user runs by hand is theirs to shape, even when it
@@ -3658,7 +3666,9 @@ def fence_gated_ports(docker_conf: dict[str, Any], env: dict[str, Any]) -> dict[
         ports[f"{int(ui['port'])}/tcp"] = bindings[0] if len(bindings) == 1 else bindings
     rng = docker_conf.get("udp_port_range")
     if isinstance(rng, dict):
-        ports.update(catalog.expand_udp_range(str(env.get(rng["env"]) or ""), int(rng.get("max_ports") or 1000)))
+        ports.update(
+            catalog.expand_udp_range(str(env.get(rng["env"]) or ""), int(rng.get("max_ports") or catalog.MAX_UDP_PORTS))
+        )
     return ports
 
 
@@ -3696,7 +3706,12 @@ async def _fence_gated_spec(
     try:
         ports = fence_gated_ports(docker_conf, env)
     except ValueError as exc:
-        raise HTTPException(status_code=400, detail=f"Invalid value for {slug}: {exc}") from None
+        # Refused before the worker is asked: a node already running keeps
+        # running as it is, whatever range it has.
+        raise HTTPException(
+            status_code=400,
+            detail=f"Invalid value for {slug}: {exc}. Nothing was deployed; a running {slug} was not touched.",
+        ) from None
     merged = dict(spec)
     # The record's network mode and ports do not win here, so saying they were kept would be false.
     divergence[:] = [d for d in divergence if not d.startswith(("network_mode: keeping", "ports: keeping"))]
@@ -4227,6 +4242,10 @@ async def _proxy_worker_command(
     # for storj). Cut off sooner, a stop that went through answers 504 and is
     # never recorded, so the copy raises a container-down alert.
     timeout = 30 + _catalog_stop_timeout(slug) if command in ("stop", "restart", "remove") else 30
+    if command == "stop" and catalog.fence_gated(catalog.get_service(slug)):
+        # The worker stops a fence-gated node under the lock a redeploy holds
+        # (orchestrator.GATED_DEPLOY_LOCK), so a Stop sent during one waits for it first.
+        timeout += _deploy_timeout(slug)
     if command == "remove":
         return await _proxy_to_worker(worker_id, "DELETE", f"/api/containers/{slug}", params=params, timeout=timeout)
     return await _proxy_to_worker(
@@ -4271,6 +4290,15 @@ def _catalog_stop_timeout(slug: str) -> int:
     return timeout if timeout > 0 else 30
 
 
+def _deploy_timeout(slug: str) -> int:
+    """How long a deploy of ``slug`` may take on the worker.
+
+    It stops the old container first, and a start Docker is slow to answer is
+    waited out for up to START_SETTLE_SECONDS (orchestrator._await_start).
+    """
+    return 60 + _catalog_stop_timeout(slug) + constants.START_SETTLE_SECONDS
+
+
 async def _proxy_worker_deploy(worker_id: int, slug: str, spec: dict[str, Any]) -> dict[str, Any]:
     """Forward a deploy command with full spec to a worker.
 
@@ -4280,7 +4308,6 @@ async def _proxy_worker_deploy(worker_id: int, slug: str, spec: dict[str, Any])
     while the worker was still shutting the old node down cleanly.
     """
     docker_conf = (catalog.get_service(slug) or {}).get("docker") or {}
-    stop_timeout = _catalog_stop_timeout(slug)
     # Which build THIS worker runs is decided here, at the edge, from the
     # architecture in its heartbeat. The deployment record stays on the
     # catalog's default image: it is one row for the whole fleet, and a
@@ -4294,7 +4321,7 @@ async def _proxy_worker_deploy(worker_id: int, slug: str, spec: dict[str, Any])
         if image:
             spec = {**spec, "image": image}
     return await _proxy_to_worker(
-        worker_id, "POST", f"/api/containers/{slug}/deploy", json=spec, timeout=60 + stop_timeout
+        worker_id, "POST", f"/api/containers/{slug}/deploy", json=spec, timeout=_deploy_timeout(slug)
     )
 
 

app/orchestrator.py

@@ -22,6 +22,7 @@
 from typing import Any
 
 import docker
+import requests
 from docker.errors import APIError, DockerException, NotFound
 
 # docker-py's run() and create() refuse stop_timeout although the Engine API and
@@ -56,6 +57,7 @@
     LABEL_SERVICE,
     LABEL_STATE_CREATED,
     LABEL_VERSION,
+    START_SETTLE_SECONDS,
     drift_warnings_enabled,
 )
 
@@ -415,6 +417,18 @@ def _fatal_entry(container: Any, slug: str) -> dict[str, Any]:
     return {"fatal": kind, "fatal_at": finished} if kind else {}
 
 
+def _held_entry(container: Any) -> dict[str, Any]:
+    """``{"held": why}`` for a container the fence guard keeps stopped (lan_fence.guard), else nothing.
+
+    The guard's last pass, so the dashboard can say on the service itself why
+    the node is down and what brings it back, not only on the Fleet page.
+    """
+    from app import lan_fence
+
+    why = (lan_fence._state.get("guard") or {}).get(getattr(container, "name", None))
+    return {"held": str(why)} if why else {}
+
+
 def _shape(container: Any) -> dict[str, Any]:
     """What a container was created with, on the settings that decide whether it works.
 
@@ -1090,17 +1104,28 @@ def deploy_raw(
         **resource_kwargs,
     )
     created = None
+    began = time.monotonic()
     try:
-        container = (
-            client.containers.create(**run_kwargs) if gated else client.containers.run(detach=True, **run_kwargs)
-        )
-        created = container
+        try:
+            container = (
+                client.containers.create(**run_kwargs) if gated else client.containers.run(detach=True, **run_kwargs)
+            )
+            created = container
+            if gated:
+                _mark_stopped(name, False)
+                # Left created, not started, when the fence is not confirmed: the guard
+                # starts it as soon as it is, and nothing runs before that.
+                _require_fence(slug)
+                container.start()
+        except ContainerNetworkError:
+            raise
+        except Exception as exc:
+            # No answer is not a refusal: Docker may still be starting it. A gated
+            # create that got none was never started, so there is nothing to wait for.
+            if not _outcome_unknown(exc) or (gated and created is None):
+                raise
+            container = created = _await_start(client, name, created, began, exc)
         if gated:
-            _mark_stopped(name, False)
-            # Left created, not started, when the fence is not confirmed: the guard
-            # starts it as soon as it is, and nothing runs before that.
-            _require_fence(slug)
-            container.start()
             from app import lan_fence
 
             if problem := lan_fence._started_unattached(container, gated_mode):
@@ -1292,6 +1317,72 @@ def _deploy_failure(exc: BaseException, restored: str, slug: str = "") -> Contai
     return ContainerRuntimeError(f"The container could not be created: {exc}. {restored}", 500)
 
 
+#: What the socket proxy answers when it stopped waiting for Docker (sockguard
+#: cuts a call whose answer has not begun after 30 s): Docker may still be at it.
+_GATEWAY_STATUSES = frozenset({502, 504})
+#: Seconds between two looks at a container whose start got no answer.
+_START_POLL_SECONDS = 2.0
+
+
+def _outcome_unknown(exc: BaseException) -> bool:
+    """Whether a Docker call that raised may still have done its work: no answer, rather than a refusal.
+
+    docker-py's own read timeout or a dropped connection, or the socket proxy's
+    gateway answer. A refusal (a taken port, a proxy rule, a bad spec) is an
+    answer, and is not waited on.
+    """
+    # APIError first: docker-py derives it from requests' HTTPError, and it is Docker's answer.
+    if isinstance(exc, APIError):
+        return getattr(exc, "status_code", None) in _GATEWAY_STATUSES
+    return isinstance(exc, requests.exceptions.RequestException)
+
+
+def _await_start(client: Any, name: str, created: Any | None, began: float, exc: BaseException) -> Any:
+    """The container under ``name`` once Docker has started it; raises when it did not.
+
+    For a create or start whose call ended without Docker's answer: Docker goes
+    on starting the container after the proxy or docker-py stopped waiting, and
+    each published port slows the start (a node with 1,000 UDP ports took about
+    51 s). Removing it then, as a refused start is, took down a node Docker was
+    still bringing up. Looks until it runs, Docker reports the start failed, or
+    START_SETTLE_SECONDS from the create call have passed.
+    """
+    logger.warning(
+        "Docker did not answer the start of %s (%s); waiting up to %d s for it to come up",
+        name,
+        exc,
+        START_SETTLE_SECONDS,
+    )
+    deadline = began + START_SETTLE_SECONDS
+    while True:
+        try:
+            current = created if created is not None else client.containers.get(name)
+            # Docker answers an inspect only once the start lets go of the container.
+            current.reload()
+        except NotFound:
+            raise ContainerRuntimeError(f"{name} is gone: Docker did not start it ({exc}).", 500) from exc
+        except Exception as poll_exc:  # noqa: BLE001 - not known yet; looked at again below
+            logger.debug("Could not read %s while it starts: %s", name, poll_exc)
+        else:
+            attrs = current.attrs if isinstance(getattr(current, "attrs", None), dict) else {}
+            state = attrs.get("State") if isinstance(attrs.get("State"), dict) else {}
+            status = getattr(current, "status", "")
+            if status == "running":
+                logger.info("%s came up %.0f s after its create call", name, time.monotonic() - began)
+                return current
+            if status in ("exited", "dead") or (status == "created" and state.get("Error")):
+                why = state.get("Error") or f"it is {status}"
+                raise ContainerRuntimeError(f"{name} did not start: {why}.", 500) from exc
+        if time.monotonic() >= deadline:
+            # 500, not 504: the deploy is rolled back by the time this is answered,
+            # and a 504 tells a caller the worker may still be at it (main keeps a
+            # proxied copy's record on one).
+            raise ContainerRuntimeError(
+                f"{name} was not running {START_SETTLE_SECONDS} s after Docker was asked to start it ({exc}).", 500
+            ) from exc
+        time.sleep(_START_POLL_SECONDS)
+
+
 from app.constants import EPHEMERAL_PORTS  # noqa: E402 - Linux's default ephemeral range
 
 
@@ -3109,6 +3200,7 @@ def get_status() -> list[dict[str, Any]]:
                 **_stop_facts(c),
                 **_state_id_entry(c, slug, client),
                 **_fatal_entry(c, slug),
+                **_held_entry(c),
             }
             # Present only when the service declares a dial-back address var;
             # the key's absence is "nothing to say", never "checked and fine".
@@ -3303,6 +3395,7 @@ def get_status_light() -> list[dict[str, Any]]:
                 **_shape_entry(c),
                 **_stop_facts(c),
                 **_fatal_entry(c, slug),
+                **_held_entry(c),
             }
             # Same contract as get_status: key present only when declared+set.
             if advertised := _advertised_address(c, slug):