fix(transport,download): harden reconnect loop and download failure paths (#372, #373)

agents/completed.md

@@ -8,6 +8,51 @@
 
 ---
 
+# #372: transport hardening — auth-failure gating, HelloAck deadline, redirect hop reset, TLS on redirects
+
+**Completed:** yes
+**Status:** DONE (2026-07-04, Claude) — fatal auth exit now requires 3 UNAUTHENTICATED strikes AND >=60s elapsed (transient hub-side rejections survivable; window resets on HelloAck); 30s deadline on the whole dial+Hello+HelloAck handshake so a stalling hub can never hang the worker (bounded attempts also make the run-loop disconnected-timeout check effective); redirect hop budget resets whenever the loop falls back to backoff, so `not_leader` routing stays alive across leadership-churn episodes; schemeless redirect targets inherit the TLS mode of the connection that issued the redirect (no plaintext downgrade) and a new `grpc_ca_bundle` setting (env `GRPC_CA_BUNDLE`) supplies a PEM root bundle for private CAs; `worker_id_mismatch`/`release_id_mismatch`/missing-identity FAILED_PRECONDITION exits immediately. 6 new tests in tests/test_worker_grpc_e2e.py.
+
+## Metadata
+- Category: bug / transport
+- Status: done
+- Passes: true
+
+## Tasks
+- [x] `_AUTH_FAILURE_EXIT_THRESHOLD=3` (transport.py:32, 260-267) fires inside ~5s of full-jitter backoff — the hub currently returns UNAUTHENTICATED for duplicate-stream and transient-pg conditions, so an asymmetric network blip kills the worker. Gate the fatal exit on true auth rejections AND elapsed time (e.g. 3 failures over >60s); tensorhub #539 fixes the status codes.
+- [x] No timeout on the HelloAck wait (`await stream.read()`, transport.py:329): a hub that accepts the stream but stalls mid-registration hangs the worker FOREVER (keepalive is answered at the h2 layer; `worker_disconnected_timeout_s` is only checked between attempts). `asyncio.wait_for(..., 30)` → ConnectionError → normal backoff; also enforce the disconnected timeout inside `_connect_once`.
+- [x] `redirect_hops` never resets after the 3-hop fallback (transport.py:270-275 vs 297-301 — reset only on successful HelloAck), permanently disabling `not_leader` routing for exactly the leadership-churn case it exists for. Reset when falling back with backoff.
+- [x] TLS is decided by URL scheme with a bare `:443` heuristic (transport.py:41-50) and redirect targets are schemeless host:port → TLS deployments redirect into plaintext dials. Inherit TLS mode from the connection that issued the redirect; add a CA-bundle setting instead of system-roots-only `grpc.ssl_channel_credentials()`.
+- [x] Exit fast on permanent FAILED_PRECONDITION (`worker_id_mismatch`/`release_id_mismatch`/missing identity) instead of burning the full disconnected-timeout (transport.py:279).
+
+## Acceptance
+Worker survives hub restarts/pg blips/leadership churn without process death; doomed configs die in seconds not minutes; TLS redirects work.
+
+---
+
+# #373: download failure path — fail fast on expired URLs, bounded verify retries, disk headroom, CAS progress, error mapping
+
+**Completed:** yes
+**Status:** DONE (2026-07-04, Claude) — typed `UrlExpiredError` (models/errors.py) raised on 4xx (except 408/429) from a presigned URL with ZERO retries; the ModelOp path emits `ModelEvent{FAILED, url_expired}` within seconds (hub re-mints fresh URLs per CONTRACT); `_is_terminal_download_error` now reads `exc.response.status_code` for requests.HTTPError (the attribute bug) — which also makes civitai 429/5xx retryable while 401/403 stay terminal INVALID; blanket 30-try/1h backoff decorator replaced with an explicit policy loop (verify size/blake3 failures capped at initial+2 retries; ENOSPC → `InsufficientDiskError` immediately; `backoff` dep dropped); pre-download disk-headroom check in `_ensure_blobs` (missing blob bytes + 1GiB headroom vs `shutil.disk_usage(...).free`) raises `InsufficientDiskError`; CAS downloads now stream `progress(bytes_done, bytes_total)` through `ensure_snapshot_async` (cached blobs pre-counted, per-chunk updates, clamped); missing orchestrator snapshot for a tensorhub ref maps to RETRYABLE instead of client-visible INVALID; job-path `UrlExpiredError` maps to RETRYABLE. 7 new tests in tests/test_download_failure_paths.py. Deferred: eviction-on-shortfall stays with #370 (headroom shortfall emits insufficient_disk immediately); no civitai-internal retry loop added — the executor's outer bounded retry covers 429/5xx now that classification is fixed.
+
+## Metadata
+- Category: bug / model-distribution
+- Status: done
+- Passes: true
+
+## Tasks
+- [x] cozy_cas.py:27-34 blanket-retries `requests.RequestException|ValueError|OSError` 30 tries/1h per file: an expired presigned URL (403, 15-min TTL hub-side) is retried against the same dead URL for up to an hour, then `_is_terminal_download_error` (executor.py:108-112) reads `exc.status_code` — which `requests.HTTPError` doesn't have (`exc.response.status_code`) — so it's classified transient and re-run 3 more outer times. Raise a typed UrlExpiredError on 4xx, emit `ModelEvent{FAILED, url_expired}` within seconds, fix the attribute bug.
+- [x] Same decorator retries blake3/size mismatches (ValueError) up to 30 full re-downloads and ENOSPC (OSError) 30 times against a full disk. Cap verify retries at 2; ENOSPC → immediate `insufficient_disk`.
+- [x] Pre-download disk-headroom check: sum missing SnapshotFile.size_bytes vs `shutil.disk_usage(cas_dir).free` with margin; on shortfall trigger the #370 eviction path or emit `insufficient_disk` immediately (`InsufficientDiskError` in capability.py:141-166 is defined but never raised).
+- [x] The CAS path never passes `progress=` to `ensure_snapshot_async` (download.py:169-173) — `bytes_done/bytes_total` never flows for tensorhub models (the main production path). Wire it.
+- [x] "tensorhub ref needs an orchestrator-resolved snapshot" (download.py:164-168) maps ValueError→INVALID→client-visible 400 for a hub-side residency bug. Map to RETRYABLE.
+- [x] civitai: no internal retry, and auth/rate-limit failures are ValueError→terminal INVALID (download.py:575); classify 429/5xx as retryable.
+
+## Acceptance
+An expired URL, a corrupted blob, and a full disk each converge in seconds with the correct CONTRACT §9 error code; CAS downloads report progress.
+
+---
+
 # #367: split clone+conversion out as `cozy_convert`; move the tenant SDK to training-endpoints
 
 **Completed:** yes

agents/progress.md

@@ -380,53 +380,6 @@ Alternating requests across two endpoints on one GPU show demote/promote transit
 
 ---
 
----
-
-# #372: transport hardening — auth-failure gating, HelloAck deadline, redirect hop reset, TLS on redirects
-
-**Completed:** no
-**Status:** OPEN (2026-07-04, gRPC connect audit; counterpart tensorhub #539)
-
-## Metadata
-- Category: bug / transport
-- Status: planned
-- Passes: false
-
-## Tasks
-- [ ] `_AUTH_FAILURE_EXIT_THRESHOLD=3` (transport.py:32, 260-267) fires inside ~5s of full-jitter backoff — the hub currently returns UNAUTHENTICATED for duplicate-stream and transient-pg conditions, so an asymmetric network blip kills the worker. Gate the fatal exit on true auth rejections AND elapsed time (e.g. 3 failures over >60s); tensorhub #539 fixes the status codes.
-- [ ] No timeout on the HelloAck wait (`await stream.read()`, transport.py:329): a hub that accepts the stream but stalls mid-registration hangs the worker FOREVER (keepalive is answered at the h2 layer; `worker_disconnected_timeout_s` is only checked between attempts). `asyncio.wait_for(..., 30)` → ConnectionError → normal backoff; also enforce the disconnected timeout inside `_connect_once`.
-- [ ] `redirect_hops` never resets after the 3-hop fallback (transport.py:270-275 vs 297-301 — reset only on successful HelloAck), permanently disabling `not_leader` routing for exactly the leadership-churn case it exists for. Reset when falling back with backoff.
-- [ ] TLS is decided by URL scheme with a bare `:443` heuristic (transport.py:41-50) and redirect targets are schemeless host:port → TLS deployments redirect into plaintext dials. Inherit TLS mode from the connection that issued the redirect; add a CA-bundle setting instead of system-roots-only `grpc.ssl_channel_credentials()`.
-- [ ] Exit fast on permanent FAILED_PRECONDITION (`worker_id_mismatch`/`release_id_mismatch`/missing identity) instead of burning the full disconnected-timeout (transport.py:279).
-
-## Acceptance
-Worker survives hub restarts/pg blips/leadership churn without process death; doomed configs die in seconds not minutes; TLS redirects work.
-
----
-
-# #373: download failure path — fail fast on expired URLs, bounded verify retries, disk headroom, CAS progress, error mapping
-
-**Completed:** no
-**Status:** OPEN (2026-07-04, download-flow audit; counterparts tensorhub #540/#541; disk-retention/eviction itself is #370, VRAM juggling is #371). Success path verified live end-to-end (tensorhub PR #73); these are the failure-path holes that reproduce the old flakiness.
-
-## Metadata
-- Category: bug / model-distribution
-- Status: planned
-- Passes: false
-
-## Tasks
-- [ ] cozy_cas.py:27-34 blanket-retries `requests.RequestException|ValueError|OSError` 30 tries/1h per file: an expired presigned URL (403, 15-min TTL hub-side) is retried against the same dead URL for up to an hour, then `_is_terminal_download_error` (executor.py:108-112) reads `exc.status_code` — which `requests.HTTPError` doesn't have (`exc.response.status_code`) — so it's classified transient and re-run 3 more outer times. Raise a typed UrlExpiredError on 4xx, emit `ModelEvent{FAILED, url_expired}` within seconds, fix the attribute bug.
-- [ ] Same decorator retries blake3/size mismatches (ValueError) up to 30 full re-downloads and ENOSPC (OSError) 30 times against a full disk. Cap verify retries at 2; ENOSPC → immediate `insufficient_disk`.
-- [ ] Pre-download disk-headroom check: sum missing SnapshotFile.size_bytes vs `shutil.disk_usage(cas_dir).free` with margin; on shortfall trigger the #370 eviction path or emit `insufficient_disk` immediately (`InsufficientDiskError` in capability.py:141-166 is defined but never raised).
-- [ ] The CAS path never passes `progress=` to `ensure_snapshot_async` (download.py:169-173) — `bytes_done/bytes_total` never flows for tensorhub models (the main production path). Wire it.
-- [ ] "tensorhub ref needs an orchestrator-resolved snapshot" (download.py:164-168) maps ValueError→INVALID→client-visible 400 for a hub-side residency bug. Map to RETRYABLE.
-- [ ] civitai: no internal retry, and auth/rate-limit failures are ValueError→terminal INVALID (download.py:575); classify 429/5xx as retryable.
-
-## Acceptance
-An expired URL, a corrupted blob, and a full disk each converge in seconds with the correct CONTRACT §9 error code; CAS downloads report progress.
-
----
-
 # #374: cozy_convert publish/clone robustness — retries, resume, ingest junk in published trees, silent empty publish
 
 **Completed:** no

pyproject.toml

@@ -28,7 +28,6 @@ dependencies = [
     "boto3>=1.41.0",
     "psutil>=7.0.0",
     "pyyaml>=6.0.0",
-    "backoff>=2.2.1",
     "blake3>=1.0.0",
     "huggingface-hub>=0.26.0",
     # endpoint.lock is TOML; msgspec.toml.encode needs tomli-w at bake time.

src/gen_worker/config/loader.py

@@ -32,6 +32,7 @@
     "HF_HOME": "hf_home",
     "TENSORHUB_PUBLIC_URL": "tensorhub_public_url",
     "ORCHESTRATOR_PUBLIC_ADDR": "orchestrator_public_addr",
+    "GRPC_CA_BUNDLE": "grpc_ca_bundle",
     "WORKER_ID": "worker_id",
     "WORKER_MODE": "worker_mode",
     "WORKER_JWT": "worker_jwt",

src/gen_worker/config/settings.py

@@ -34,6 +34,9 @@ class Settings(msgspec.Struct, frozen=True, kw_only=True):
     # Single shared value across the cluster — the router resolves us to the
     # per-release lease owner. gen-orchestrator issue #317.
     orchestrator_public_addr: str = ""
+    # Optional PEM CA bundle for the orchestrator gRPC TLS connection.
+    # Empty = system roots.
+    grpc_ca_bundle: str = ""
 
     # Per-pod worker identity (orchestrator-injected).
     worker_id: str = ""

src/gen_worker/executor.py

@@ -28,10 +28,11 @@
 )
 from .api.streaming import BatchItemDelta, Done, Error, IncrementalTokenDelta
 from .api.types import Compute
-from .capability import HardwareUnmetError
+from .capability import HardwareUnmetError, InsufficientDiskError
 from .models import residency as residency_mod
 from .models.cache_paths import tensorhub_cas_dir
 from .models.download import ensure_local
+from .models.errors import UrlExpiredError
 from .models.residency import Residency
 from .pb import worker_scheduler_pb2 as pb
 from .registry import EndpointSpec
@@ -83,6 +84,9 @@ def _map_exception(exc: BaseException) -> Tuple[int, str]:
         return pb.JOB_STATUS_RETRYABLE, _sanitize(str(exc) or "artifact transfer failed")
     if isinstance(exc, HardwareUnmetError):
         return pb.JOB_STATUS_RETRYABLE, _sanitize(str(exc) or "hardware unmet")
+    if isinstance(exc, UrlExpiredError):
+        # Hub-side URL staleness, not a client problem — retry re-mints URLs.
+        return pb.JOB_STATUS_RETRYABLE, "model download url expired"
     if type(exc).__name__ in ("OutOfMemoryError", "CUDAOutOfMemoryError"):
         return pb.JOB_STATUS_RETRYABLE, "out of memory"
     return pb.JOB_STATUS_FATAL, "internal error"
@@ -116,8 +120,13 @@ def _model_op_error_vocab(exc: BaseException) -> str:
 
 
 def _is_terminal_download_error(exc: BaseException) -> bool:
+    if isinstance(exc, (UrlExpiredError, InsufficientDiskError)):
+        return True
     status = getattr(exc, "status_code", None)
-    if isinstance(status, int) and 400 <= status < 500 and status != 429:
+    if not isinstance(status, int):
+        # requests.HTTPError carries the code on .response, not the exception.
+        status = getattr(getattr(exc, "response", None), "status_code", None)
+    if isinstance(status, int) and 400 <= status < 500 and status not in (408, 429):
         return True
     return isinstance(exc, (ValueError, KeyError))
 
@@ -269,6 +278,10 @@ def _progress(done: int, total: Optional[int]) -> None:
 
     @staticmethod
     def _error_vocab(exc: BaseException) -> str:
+        if isinstance(exc, UrlExpiredError):
+            return "url_expired"
+        if isinstance(exc, InsufficientDiskError):
+            return "insufficient_disk"
         text = str(exc).lower()
         if "expired" in text or "403" in text:
             return "url_expired"

src/gen_worker/models/cozy_cas.py

@@ -1,18 +1,31 @@
 from __future__ import annotations
 
 import asyncio
+import errno
 import logging
+import random
+import time
 from pathlib import Path
-from typing import Dict
+from typing import Callable, Dict, Optional
 
-import backoff
 import requests
 from blake3 import blake3
 
+from ..capability import InsufficientDiskError
+from .errors import UrlExpiredError
+
 _log = logging.getLogger(__name__)
 
 _DOWNLOAD_CHUNK_BYTES = 4 * 1024 * 1024
 
+# Retry policy: transient network errors get a long budget; verification
+# failures (size/blake3 mismatch) mean the source blob is likely corrupt —
+# 2 re-downloads, then give up. UrlExpiredError / ENOSPC never retry.
+_TRANSIENT_MAX_TRIES = 30
+_MAX_RETRY_TIME_S = 3600.0
+_VERIFY_MAX_FAILURES = 3  # initial try + 2 retries
+_RETRY_BACKOFF_CAP_S = 30.0
+
 
 def _norm_rel_path(p: str) -> str:
     s = (p or "").strip().replace("\\", "/")
@@ -24,22 +37,62 @@ def _norm_rel_path(p: str) -> str:
     return s
 
 
-@backoff.on_exception(
-    backoff.expo,
-    (requests.RequestException, ValueError, OSError),
-    max_tries=30,
-    max_time=3600,
-    factor=1,
-    max_value=30,  # cap backoff at 30s between retries
-)
-async def _download_one_file(url: str, dst: Path, expected_size: int, expected_blake3: str) -> None:
-    await asyncio.get_running_loop().run_in_executor(
-        None,
-        lambda: _download_one_file_sync(url, dst, expected_size, expected_blake3),
-    )
-
-
-def _download_one_file_sync(url: str, dst: Path, expected_size: int, expected_blake3: str) -> None:
+def _check_status(resp: requests.Response) -> None:
+    """4xx (except 408/429) on a presigned URL is dead — no retry can help."""
+    sc = int(resp.status_code)
+    if 400 <= sc < 500 and sc not in (408, 429):
+        raise UrlExpiredError(f"download URL rejected with HTTP {sc}", status_code=sc)
+    resp.raise_for_status()
+
+
+async def _download_one_file(
+    url: str,
+    dst: Path,
+    expected_size: int,
+    expected_blake3: str,
+    on_bytes: Optional[Callable[[int], None]] = None,
+) -> None:
+    loop = asyncio.get_running_loop()
+    started = time.monotonic()
+    transient_failures = 0
+    verify_failures = 0
+    while True:
+        try:
+            await loop.run_in_executor(
+                None,
+                lambda: _download_one_file_sync(
+                    url, dst, expected_size, expected_blake3, on_bytes=on_bytes
+                ),
+            )
+            return
+        except (UrlExpiredError, InsufficientDiskError):
+            raise
+        except (requests.RequestException, ValueError, OSError) as exc:
+            if getattr(exc, "errno", None) == errno.ENOSPC:
+                raise InsufficientDiskError(
+                    f"insufficient disk while downloading {dst.name}", path=str(dst.parent)
+                ) from exc
+            if isinstance(exc, ValueError):
+                verify_failures += 1
+                failures, cap = verify_failures, _VERIFY_MAX_FAILURES
+            else:
+                transient_failures += 1
+                failures, cap = transient_failures, _TRANSIENT_MAX_TRIES
+            if failures >= cap or time.monotonic() - started >= _MAX_RETRY_TIME_S:
+                raise
+            delay = random.uniform(0.5, 1.0) * min(_RETRY_BACKOFF_CAP_S, 2.0 ** failures)
+            _log.warning("download of %s failed (%s, failure %d/%d): %s; retrying in %.1fs",
+                         dst.name, type(exc).__name__, failures, cap, exc, delay)
+            await asyncio.sleep(delay)
+
+
+def _download_one_file_sync(
+    url: str,
+    dst: Path,
+    expected_size: int,
+    expected_blake3: str,
+    on_bytes: Optional[Callable[[int], None]] = None,
+) -> None:
     """Download a single file with HTTP Range resume, size + blake3 validation.
 
     Caller runs this in a worker thread; the transfer itself uses the same
@@ -120,6 +173,8 @@ def _stream(resp: requests.Response, *, write_mode: str, start: int) -> None:
                     continue
                 f.write(chunk)
                 downloaded += len(chunk)
+                if on_bytes is not None:
+                    on_bytes(len(chunk))
                 if expected_size and downloaded > expected_size:
                     raise ValueError(f"download exceeded expected size ({downloaded} > {expected_size})")
                 if downloaded - last_log >= log_every:
@@ -141,10 +196,10 @@ def _stream(resp: requests.Response, *, write_mode: str, start: int) -> None:
                 log.info("download_range_ignored path=%s status=%s (restarting from 0)", dst.name, resp.status_code)
                 resp.close()
                 with session.get(url, stream=True, timeout=(60, 180)) as resp2:
-                    resp2.raise_for_status()
+                    _check_status(resp2)
                     _stream(resp2, write_mode="wb", start=0)
             else:
-                resp.raise_for_status()
+                _check_status(resp)
                 _stream(resp, write_mode=mode, start=offset)
 
     # Validate.