diff --git a/agents/completed.md b/agents/completed.md index 77c8df7..c8cbf20 100644 --- a/agents/completed.md +++ b/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 diff --git a/agents/progress.md b/agents/progress.md index ee0e403..32ab7a0 100644 --- a/agents/progress.md +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 9e80b0b..38bd0c5 100644 --- a/pyproject.toml +++ b/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. diff --git a/src/gen_worker/config/loader.py b/src/gen_worker/config/loader.py index 68a0f5e..a047516 100644 --- a/src/gen_worker/config/loader.py +++ b/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", diff --git a/src/gen_worker/config/settings.py b/src/gen_worker/config/settings.py index 058a06a..25cf5a7 100644 --- a/src/gen_worker/config/settings.py +++ b/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 = "" diff --git a/src/gen_worker/executor.py b/src/gen_worker/executor.py index d84a197..a8f2e17 100644 --- a/src/gen_worker/executor.py +++ b/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" diff --git a/src/gen_worker/models/cozy_cas.py b/src/gen_worker/models/cozy_cas.py index 18d2133..c4224ad 100644 --- a/src/gen_worker/models/cozy_cas.py +++ b/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. diff --git a/src/gen_worker/models/cozy_snapshot.py b/src/gen_worker/models/cozy_snapshot.py index c56dee7..73d007d 100644 --- a/src/gen_worker/models/cozy_snapshot.py +++ b/src/gen_worker/models/cozy_snapshot.py @@ -8,16 +8,22 @@ import shutil import threading from pathlib import Path -from typing import Any, Dict, List, Optional, Set +from typing import Any, Callable, Dict, List, Optional, Set from .cozy_cas import _download_one_file as _download_one_file from .cozy_cas import _norm_rel_path from .hub_client import WorkerResolvedRepo, WorkerResolvedRepoFile from .refs import TensorhubRef +from ..capability import InsufficientDiskError from ..s3_transfer import S3TransferGrant, download_file_with_grant _log = logging.getLogger("gen_worker.download") +ProgressFn = Callable[[int, Optional[int]], None] + +# Free space that must remain after downloading the missing blobs. +_DISK_HEADROOM_BYTES = 1 << 30 + # --------------------------------------------------------------------------- # Snapshot build coordination (threading-based, works across event loops) @@ -160,6 +166,7 @@ async def ensure_snapshot( ref: TensorhubRef, *, resolved: Any, + progress: Optional[ProgressFn] = None, ) -> Path: blobs_root = base_dir / "blobs" snaps_root = base_dir / "snapshots" @@ -201,7 +208,7 @@ async def ensure_snapshot( try: _log.info("snapshot_build_start digest=%s files=%d", res.snapshot_digest[:16], len(res.files)) - await self._ensure_blobs(blobs_root, res.files) + await self._ensure_blobs(blobs_root, res.files, progress=progress) tmp = snaps_root / f"{res.snapshot_digest}.building" if tmp.exists(): @@ -241,7 +248,13 @@ async def ensure_snapshot( # Blob download (deduplicated, parallel) # ------------------------------------------------------------------ - async def _ensure_blobs(self, blobs_root: Path, files: List[WorkerResolvedRepoFile]) -> None: + async def _ensure_blobs( + self, + blobs_root: Path, + files: List[WorkerResolvedRepoFile], + *, + progress: Optional[ProgressFn] = None, + ) -> None: # Deduplicate by digest — same blob referenced by multiple paths (e.g. # fp16 and normal variants sharing the same part) is downloaded once. seen: Set[str] = set() @@ -258,6 +271,37 @@ async def _ensure_blobs(self, blobs_root: Path, files: List[WorkerResolvedRepoFi _log.info("ensure_blobs total_entries=%d unique_blobs=%d", len(files), len(unique)) + cached_digests = { + f.blake3.strip().lower() for f in unique + if _blob_path(blobs_root, f.blake3.strip().lower()).exists() + } + missing_bytes = sum( + int(f.size_bytes or 0) for f in unique + if f.blake3.strip().lower() not in cached_digests + ) + self._check_disk_headroom(blobs_root, missing_bytes) + + total = sum(int(f.size_bytes or 0) for f in unique) or None + done = total - missing_bytes if total else 0 + done_lock = threading.Lock() + + def _on_bytes(n: int) -> None: + nonlocal done + with done_lock: + done += n + d = done if total is None else min(done, total) + if progress is not None: + try: + progress(d, total) + except Exception: + pass + + if progress is not None: + try: + progress(min(done, total) if total else done, total) + except Exception: + pass + # Sort largest first for better overlap, then download in parallel. unique.sort(key=lambda f: int(f.size_bytes or 0), reverse=True) @@ -287,6 +331,7 @@ async def _dl(f: WorkerResolvedRepoFile) -> None: expected_blake3=digest, ), ) + _on_bytes(int(f.size_bytes or 0)) else: assert f.url is not None # validated above in _ensure_blobs loop await _download_one_file( @@ -294,11 +339,30 @@ async def _dl(f: WorkerResolvedRepoFile) -> None: dst, expected_size=int(f.size_bytes or 0), expected_blake3=digest, + on_bytes=_on_bytes, ) _log.info("blob_download_done path=%s digest=%s", f.path, digest[:16]) await asyncio.gather(*(_dl(f) for f in unique)) + @staticmethod + def _check_disk_headroom(blobs_root: Path, missing_bytes: int) -> None: + if missing_bytes <= 0: + return + try: + free = shutil.disk_usage(blobs_root).free + except OSError: + return + required = missing_bytes + _DISK_HEADROOM_BYTES + if required > free: + raise InsufficientDiskError( + f"insufficient disk for snapshot download: need {required} bytes " + f"({missing_bytes} blobs + headroom), {free} free at {blobs_root}", + available_bytes=free, + required_bytes=required, + path=str(blobs_root), + ) + # ------------------------------------------------------------------ # Chunked file reassembly # ------------------------------------------------------------------ @@ -371,6 +435,7 @@ async def ensure_snapshot_async( base_dir: Path, ref: TensorhubRef, resolved: Any, + progress: Optional[ProgressFn] = None, ) -> Path: dl = CozySnapshotDownloader() - return await dl.ensure_snapshot(base_dir, ref, resolved=resolved) + return await dl.ensure_snapshot(base_dir, ref, resolved=resolved, progress=progress) diff --git a/src/gen_worker/models/download.py b/src/gen_worker/models/download.py index fbf083e..2c8a35a 100644 --- a/src/gen_worker/models/download.py +++ b/src/gen_worker/models/download.py @@ -162,14 +162,18 @@ async def ensure_local( if parsed.provider == "tensorhub" and parsed.tensorhub is not None: if snapshot is None: - raise ValueError( + # Hub-side residency bug (the orchestrator failed to pre-resolve), + # not bad client input — RETRYABLE, never a client-visible 400. + from ..api.errors import RetryableError + + raise RetryableError( f"tensorhub ref {ref!r} needs an orchestrator-resolved snapshot " "and none was provided" ) from .cozy_snapshot import ensure_snapshot_async return await ensure_snapshot_async( - base_dir=base, ref=parsed.tensorhub, resolved=snapshot, + base_dir=base, ref=parsed.tensorhub, resolved=snapshot, progress=progress, ) if parsed.provider == "hf" and parsed.hf is not None: diff --git a/src/gen_worker/models/errors.py b/src/gen_worker/models/errors.py new file mode 100644 index 0000000..a12a7eb --- /dev/null +++ b/src/gen_worker/models/errors.py @@ -0,0 +1,15 @@ +"""Typed download-failure errors (CONTRACT §9 ModelEvent.error vocabulary).""" +from __future__ import annotations + + +class UrlExpiredError(RuntimeError): + """A presigned download URL was rejected with a permanent 4xx (expired + signature, revoked object). Never retried worker-side: the orchestrator + re-mints fresh URLs on ``ModelEvent{FAILED, error:"url_expired"}``.""" + + def __init__(self, message: str, *, status_code: int = 0) -> None: + super().__init__(message) + self.status_code = int(status_code) + + +__all__ = ["UrlExpiredError"] diff --git a/src/gen_worker/transport.py b/src/gen_worker/transport.py index c748eaf..28630cd 100644 --- a/src/gen_worker/transport.py +++ b/src/gen_worker/transport.py @@ -30,7 +30,19 @@ _MAX_REDIRECT_HOPS = 3 _AUTH_FAILURE_EXIT_THRESHOLD = 3 +# UNAUTHENTICATED can be transient hub-side (duplicate stream teardown, pg +# blip): exit only when failures persist across a real time window. +_AUTH_FAILURE_EXIT_WINDOW_S = 60.0 _BACKOFF_RESET_AFTER_S = 60.0 +_HELLO_ACK_TIMEOUT_S = 30.0 +# FAILED_PRECONDITION details that can never heal by retrying: identity is +# wrong for this deployment, so exit for reap instead of burning the full +# disconnected timeout. +_PERMANENT_PRECONDITION_MARKERS = ( + "worker_id_mismatch", + "release_id_mismatch", + "missing worker identity", +) class FatalTransportError(Exception): @@ -38,8 +50,13 @@ class FatalTransportError(Exception): disconnected-timeout elapsed. The worker process should exit.""" -def normalize_grpc_addr(addr: str) -> Tuple[str, bool]: - """Normalize a scheduler address into (host:port, use_tls).""" +def normalize_grpc_addr(addr: str, default_tls: Optional[bool] = None) -> Tuple[str, bool]: + """Normalize a scheduler address into (host:port, use_tls). + + ``default_tls`` is the TLS mode for schemeless addresses (e.g. a + ``not_leader`` redirect target inherits it from the connection that issued + the redirect); when None, fall back to the bare ``:443`` heuristic. + """ a = (addr or "").strip() if not a: return "", False @@ -47,6 +64,8 @@ def normalize_grpc_addr(addr: str) -> Tuple[str, bool]: for prefix, tls in (("grpcs://", True), ("grpc://", False), ("https://", True), ("http://", False)): if lower.startswith(prefix): return a[len(prefix):].strip(), tls + if default_tls is not None: + return a, default_tls return a, a.endswith(":443") @@ -185,6 +204,7 @@ def __init__( self._clean_close = False self.reconnect_delays: List[float] = [] # observability + tests self._consecutive_auth_failures = 0 + self._first_auth_failure_at: Optional[float] = None self._connected_at: Optional[float] = None # set on each HelloAck # ---- send API -------------------------------------------------------- @@ -227,11 +247,17 @@ def _channel_options(self) -> List[Tuple[str, int]]: ("grpc.max_receive_message_length", 64 * 1024 * 1024), ] - def _make_channel(self, addr: str) -> grpc.aio.Channel: - target, use_tls = normalize_grpc_addr(addr) + def _make_channel(self, target: str, use_tls: bool) -> grpc.aio.Channel: if use_tls: + roots: Optional[bytes] = None + bundle = (getattr(self._settings, "grpc_ca_bundle", "") or "").strip() + if bundle: + with open(bundle, "rb") as f: + roots = f.read() return grpc.aio.secure_channel( - target, grpc.ssl_channel_credentials(), options=self._channel_options() + target, + grpc.ssl_channel_credentials(root_certificates=roots), + options=self._channel_options(), ) return grpc.aio.insecure_channel(target, options=self._channel_options()) @@ -246,35 +272,52 @@ async def run(self) -> None: FatalTransportError on version mismatch / auth lockout / timeout.""" attempt = 0 redirect_addr: Optional[str] = None + redirect_tls: Optional[bool] = None redirect_hops = 0 last_connected = time.monotonic() while not self._stopping.is_set(): - addr = redirect_addr or self._settings.orchestrator_public_addr - redirect_addr = None + if redirect_addr is not None: + # Schemeless redirect targets inherit the TLS mode of the + # connection that issued the redirect — never downgrade. + target, use_tls = normalize_grpc_addr(redirect_addr, default_tls=redirect_tls) + redirect_addr = None + else: + target, use_tls = normalize_grpc_addr(self._settings.orchestrator_public_addr) self._connected_at = None try: - await self._connect_once(addr) + await self._connect_once(target, use_tls) except grpc.aio.AioRpcError as e: code, details = e.code(), str(e.details() or "") if code == grpc.StatusCode.UNAUTHENTICATED: + now = time.monotonic() + if self._first_auth_failure_at is None: + self._first_auth_failure_at = now self._consecutive_auth_failures += 1 - logger.error("stream rejected UNAUTHENTICATED (%d consecutive): %s", - self._consecutive_auth_failures, details) - if self._consecutive_auth_failures >= _AUTH_FAILURE_EXIT_THRESHOLD: + logger.error("stream rejected UNAUTHENTICATED (%d consecutive over %.0fs): %s", + self._consecutive_auth_failures, + now - self._first_auth_failure_at, details) + if ( + self._consecutive_auth_failures >= _AUTH_FAILURE_EXIT_THRESHOLD + and now - self._first_auth_failure_at >= _AUTH_FAILURE_EXIT_WINDOW_S + ): raise FatalTransportError( - f"authentication rejected {self._consecutive_auth_failures} times: {details}" + f"authentication rejected {self._consecutive_auth_failures} times " + f"over {now - self._first_auth_failure_at:.0f}s: {details}" ) from e elif code == grpc.StatusCode.FAILED_PRECONDITION: if details.startswith("not_leader:"): if redirect_hops < _MAX_REDIRECT_HOPS: redirect_hops += 1 redirect_addr = details.split(":", 1)[1].strip() + redirect_tls = use_tls logger.info("not_leader redirect -> %s (hop %d)", redirect_addr, redirect_hops) continue # immediate, no backoff logger.warning("redirect hop limit reached; falling back with backoff") elif "protocol_version_mismatch" in details: raise FatalTransportError(f"protocol version mismatch: {details}") from e + elif any(m in details for m in _PERMANENT_PRECONDITION_MARKERS): + raise FatalTransportError(f"permanent registration rejection: {details}") from e else: logger.error("protocol violation: %s", details) else: @@ -282,7 +325,7 @@ async def run(self) -> None: except FatalTransportError: raise except Exception as e: - logger.warning("connection to %s failed: %s: %s", addr, type(e).__name__, e) + logger.warning("connection to %s failed: %s: %s", target, type(e).__name__, e) finally: self._connected.clear() await self.queue.reset_for_reconnect() @@ -294,10 +337,13 @@ async def run(self) -> None: if self._stopping.is_set(): return + # The immediate-redirect chain is over; the hop budget refreshes + # for the next leadership-churn episode. + redirect_hops = 0 + now = time.monotonic() if self._connected_at is not None: last_connected = now - redirect_hops = 0 if now - self._connected_at >= _BACKOFF_RESET_AFTER_S: attempt = 0 timeout_s = float(self._settings.worker_disconnected_timeout_s or 0) @@ -316,17 +362,26 @@ async def run(self) -> None: except asyncio.TimeoutError: pass - async def _connect_once(self, addr: str) -> None: + async def _connect_once(self, target: str, use_tls: bool) -> None: """One connection lifetime; sets self._connected_at once HelloAck lands.""" - channel = self._make_channel(addr) + channel = self._make_channel(target, use_tls) try: stub = pb_grpc.WorkerSchedulerStub(channel) stream = stub.Connect(metadata=self._metadata()) - hello = self._handlers.build_hello() - await stream.write(pb.WorkerMessage(hello=hello)) + async def _handshake() -> Any: + await stream.write(pb.WorkerMessage(hello=self._handlers.build_hello())) + return await stream.read() - first = await stream.read() + # Deadline on the whole dial+Hello+HelloAck handshake: a hub that + # accepts the stream but never answers must not hang the worker + # forever (h2 keepalive is answered below the app layer). + try: + first = await asyncio.wait_for(_handshake(), _HELLO_ACK_TIMEOUT_S) + except asyncio.TimeoutError: + raise ConnectionError( + f"no HelloAck within {_HELLO_ACK_TIMEOUT_S:.0f}s" + ) from None if first is grpc.aio.EOF: raise ConnectionError("stream closed before HelloAck") if first.WhichOneof("msg") != "hello_ack": @@ -339,8 +394,9 @@ async def _connect_once(self, addr: str) -> None: ) self._connected_at = time.monotonic() self._consecutive_auth_failures = 0 + self._first_auth_failure_at = None self._connected.set() - logger.info("connected to %s (HelloAck ok)", addr) + logger.info("connected to %s (HelloAck ok)", target) await self._handlers.on_hello_ack(ack) send_task = asyncio.create_task(self._send_loop(stream), name="transport-send") diff --git a/tests/test_cozy_snapshot.py b/tests/test_cozy_snapshot.py index 163b9e2..9423120 100644 --- a/tests/test_cozy_snapshot.py +++ b/tests/test_cozy_snapshot.py @@ -35,7 +35,7 @@ def _resolved() -> dict: def test_failed_build_is_evicted_and_retry_succeeds(tmp_path: Path, monkeypatch) -> None: calls = {"n": 0} - async def _flaky_download(url: str, dst: Path, expected_size: int, expected_blake3: str) -> None: + async def _flaky_download(url: str, dst: Path, expected_size: int, expected_blake3: str, on_bytes=None) -> None: calls["n"] += 1 if calls["n"] == 1: raise RuntimeError("transient blob failure") diff --git a/tests/test_download_failure_paths.py b/tests/test_download_failure_paths.py new file mode 100644 index 0000000..36bda85 --- /dev/null +++ b/tests/test_download_failure_paths.py @@ -0,0 +1,184 @@ +"""Download failure paths (#373): expired presigned URLs fail in seconds with +ModelEvent{FAILED, url_expired} (the hub re-mints fresh URLs), verify +mismatches get a bounded retry budget, a full disk surfaces as +insufficient_disk immediately, and CAS downloads report byte progress.""" + +from __future__ import annotations + +import asyncio +import errno +import http.server +import threading +from pathlib import Path + +import pytest +import requests + +import gen_worker.models.cozy_cas as cas_mod +import gen_worker.models.cozy_snapshot as snap_mod +from blake3 import blake3 +from gen_worker.capability import InsufficientDiskError +from gen_worker.executor import ModelStore, _is_terminal_download_error +from gen_worker.models.cozy_cas import _download_one_file +from gen_worker.models.cozy_snapshot import ensure_snapshot_async +from gen_worker.models.errors import UrlExpiredError +from gen_worker.models.refs import TensorhubRef +from gen_worker.pb import worker_scheduler_pb2 as pb + + +def _serve(handler_cls) -> tuple[http.server.ThreadingHTTPServer, str]: + httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler_cls) + threading.Thread(target=httpd.serve_forever, daemon=True).start() + return httpd, f"http://127.0.0.1:{httpd.server_address[1]}" + + +class _CountingHandler(http.server.BaseHTTPRequestHandler): + hits = 0 + payload = b"" + status = 200 + + def do_GET(self): # noqa: N802 + type(self).hits += 1 + self.send_response(self.status) + self.send_header("Content-Length", str(len(self.payload))) + self.end_headers() + if self.payload: + self.wfile.write(self.payload) + + def log_message(self, *a): + pass + + +def test_expired_url_fails_immediately_without_retries(tmp_path: Path) -> None: + class _Expired(_CountingHandler): + hits = 0 + status = 403 + + httpd, base = _serve(_Expired) + try: + with pytest.raises(UrlExpiredError) as ei: + asyncio.run(_download_one_file( + f"{base}/blob", tmp_path / "blob", expected_size=4, expected_blake3="")) + assert ei.value.status_code == 403 + assert _Expired.hits == 1 # dead URL: zero retries + finally: + httpd.shutdown() + + +def test_verify_mismatch_retries_are_bounded(tmp_path: Path) -> None: + class _Corrupt(_CountingHandler): + hits = 0 + payload = b"garbage!" + + httpd, base = _serve(_Corrupt) + try: + with pytest.raises(ValueError, match="blake3 mismatch"): + asyncio.run(_download_one_file( + f"{base}/blob", tmp_path / "blob", + expected_size=len(_Corrupt.payload), + expected_blake3=blake3(b"expected-content").hexdigest())) + assert _Corrupt.hits == cas_mod._VERIFY_MAX_FAILURES # initial + 2 retries + finally: + httpd.shutdown() + + +def test_enospc_raises_insufficient_disk_immediately(tmp_path: Path, monkeypatch) -> None: + calls = {"n": 0} + + def _full_disk(*a, **kw): + calls["n"] += 1 + raise OSError(errno.ENOSPC, "No space left on device") + + monkeypatch.setattr(cas_mod, "_download_one_file_sync", _full_disk) + with pytest.raises(InsufficientDiskError): + asyncio.run(_download_one_file( + "http://example.invalid/blob", tmp_path / "blob", + expected_size=4, expected_blake3="")) + assert calls["n"] == 1 + + +def test_terminal_download_error_classification() -> None: + def _http_error(status: int) -> requests.HTTPError: + resp = requests.Response() + resp.status_code = status + return requests.HTTPError(f"HTTP {status}", response=resp) + + assert _is_terminal_download_error(UrlExpiredError("gone", status_code=403)) + assert _is_terminal_download_error(InsufficientDiskError("full")) + assert _is_terminal_download_error(_http_error(403)) # exc.response.status_code (#373) + assert _is_terminal_download_error(_http_error(404)) + assert not _is_terminal_download_error(_http_error(429)) + assert not _is_terminal_download_error(_http_error(408)) + assert not _is_terminal_download_error(_http_error(500)) + assert not _is_terminal_download_error(requests.ConnectionError("reset")) + + +def _resolved(payload: bytes, url: str) -> dict: + return { + "snapshot_digest": "ab" * 32, + "files": [{ + "path": "model.safetensors", + "size_bytes": len(payload), + "blake3": blake3(payload).hexdigest(), + "url": url, + }], + } + + +def test_snapshot_disk_headroom_check(tmp_path: Path, monkeypatch) -> None: + class _Usage: + free = 1 # byte + + monkeypatch.setattr(snap_mod.shutil, "disk_usage", lambda p: _Usage) + with pytest.raises(InsufficientDiskError): + asyncio.run(ensure_snapshot_async( + base_dir=tmp_path, ref=TensorhubRef(owner="e2e", repo="tiny"), + resolved=_resolved(b"12345", "http://example.invalid/blob"))) + + +def test_snapshot_download_reports_progress(tmp_path: Path) -> None: + payload = b"x" * 5000 + + class _Blob(_CountingHandler): + hits = 0 + + _Blob.payload = payload + httpd, base = _serve(_Blob) + seen: list[tuple[int, int | None]] = [] + try: + out = asyncio.run(ensure_snapshot_async( + base_dir=tmp_path, ref=TensorhubRef(owner="e2e", repo="tiny"), + resolved=_resolved(payload, f"{base}/blob"), + progress=lambda done, total: seen.append((done, total)))) + assert (out / "model.safetensors").read_bytes() == payload + assert seen and seen[-1] == (len(payload), len(payload)) + finally: + httpd.shutdown() + + +def test_model_store_emits_url_expired_within_one_attempt(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("TENSORHUB_CACHE_DIR", str(tmp_path)) + + class _Expired(_CountingHandler): + hits = 0 + status = 403 + + httpd, base = _serve(_Expired) + sent: list[pb.WorkerMessage] = [] + + async def _emit(msg: pb.WorkerMessage) -> None: + sent.append(msg) + + snapshot = pb.Snapshot(digest="snap-1", files=[pb.SnapshotFile( + path="model.safetensors", size_bytes=4, blake3="cd" * 32, url=f"{base}/blob")]) + try: + store = ModelStore(_emit, cache_dir=tmp_path) + with pytest.raises(UrlExpiredError): + asyncio.run(store.ensure_local("e2e/tiny", snapshot)) + assert _Expired.hits == 1 # no outer executor retries either + failed = [m.model_event for m in sent + if m.WhichOneof("msg") == "model_event" + and m.model_event.state == pb.MODEL_STATE_FAILED] + assert failed and failed[-1].error == "url_expired" + finally: + httpd.shutdown() diff --git a/tests/test_provider_routing.py b/tests/test_provider_routing.py index a2e6ba2..b49b8f6 100644 --- a/tests/test_provider_routing.py +++ b/tests/test_provider_routing.py @@ -179,11 +179,14 @@ def _fake_civitai(version_id, out_dir, **kw): def test_tensorhub_refs_require_a_snapshot(tmp_path: Path, monkeypatch, ref, index) -> None: """Workers never resolve tensorhub refs themselves — the orchestrator pre-resolves and ships a Snapshot. Without one, the tensorhub branch - raises (and the HF branch is NOT touched).""" + raises RETRYABLE (a hub residency bug, not client input; #373) and the + HF branch is NOT touched.""" + from gen_worker.api.errors import RetryableError + calls: list = [] monkeypatch.setattr(dl_mod, "download_hf", lambda *a, **k: calls.append(a) or tmp_path) set_provider_index(index) - with pytest.raises(ValueError, match="orchestrator-resolved snapshot"): + with pytest.raises(RetryableError, match="orchestrator-resolved snapshot"): asyncio.run(ensure_local(ref, cache_dir=tmp_path)) assert calls == [] diff --git a/tests/test_worker_grpc_e2e.py b/tests/test_worker_grpc_e2e.py index 21eae00..7df382f 100644 --- a/tests/test_worker_grpc_e2e.py +++ b/tests/test_worker_grpc_e2e.py @@ -434,7 +434,13 @@ def test_marco_polo_example_serves_under_the_new_core() -> None: sys.path.remove(str(src)) -def test_auth_rejection_exits_instead_of_spinning() -> None: +def test_auth_rejection_exits_instead_of_spinning(monkeypatch) -> None: + # UNAUTHENTICATED can be transient hub-side; the fatal exit is gated on + # BOTH a failure count and an elapsed window (#372). Shrink the window so + # the test observes the exit quickly. + import gen_worker.transport as transport_mod + + monkeypatch.setattr(transport_mod, "_AUTH_FAILURE_EXIT_WINDOW_S", 0.3) scheduler = FakeScheduler(reject_unauthenticated=True) server = grpc.server(futures.ThreadPoolExecutor(max_workers=4)) pb_grpc.add_WorkerSchedulerServicer_to_server(scheduler, server) @@ -448,6 +454,131 @@ def test_auth_rejection_exits_instead_of_spinning() -> None: server.stop(grace=0) +def test_auth_rejection_within_window_keeps_retrying() -> None: + """3 quick UNAUTHENTICATED strikes must NOT kill the worker while the + exit window has not elapsed — a hub pg blip is survivable (#372).""" + scheduler = FakeScheduler(reject_unauthenticated=True) + server = grpc.server(futures.ThreadPoolExecutor(max_workers=4)) + pb_grpc.add_WorkerSchedulerServicer_to_server(scheduler, server) + port = server.add_insecure_port("127.0.0.1:0") + server.start() + try: + harness = _Harness(scheduler, port) + harness.start() + deadline = time.monotonic() + 3.0 + while len(harness.worker.transport.reconnect_delays) < 4: + assert harness._thread.is_alive(), "worker exited inside the auth window" + assert time.monotonic() < deadline, "worker never retried" + time.sleep(0.02) + assert harness.exit_code is None + finally: + harness.stop() + server.stop(grace=0) + + +def test_permanent_precondition_exits_fast() -> None: + """worker_id_mismatch cannot heal by retrying: exit for reap immediately + instead of burning the disconnected timeout (#372).""" + + class _Mismatch(FakeScheduler): + def Connect(self, request_iterator, context): + context.abort(grpc.StatusCode.FAILED_PRECONDITION, + "worker_id_mismatch: hello=w1 jwt_sub=w2") + + scheduler = _Mismatch() + server = grpc.server(futures.ThreadPoolExecutor(max_workers=4)) + pb_grpc.add_WorkerSchedulerServicer_to_server(scheduler, server) + port = server.add_insecure_port("127.0.0.1:0") + server.start() + try: + harness = _Harness(scheduler, port) + harness.start() + assert harness.join(timeout=_TIMEOUT) == 1 + finally: + server.stop(grace=0) + + +def test_hello_ack_deadline_reconnects_instead_of_hanging(monkeypatch) -> None: + """A hub that accepts the stream but never sends HelloAck must not hang + the worker forever (#372): the handshake deadline fires and the worker + reconnects with backoff.""" + import gen_worker.transport as transport_mod + + monkeypatch.setattr(transport_mod, "_HELLO_ACK_TIMEOUT_S", 0.3) + + stalls = {"n": 0} + + class _Stall(FakeScheduler): + def Connect(self, request_iterator, context): + stalls["n"] += 1 + if stalls["n"] == 1: + next(request_iterator) # read Hello, then say nothing + time.sleep(5.0) # stall well past the deadline + return iter(()) + return super().Connect(request_iterator, context) + + scheduler = _Stall() + server = grpc.server(futures.ThreadPoolExecutor(max_workers=4)) + pb_grpc.add_WorkerSchedulerServicer_to_server(scheduler, server) + port = server.add_insecure_port("127.0.0.1:0") + server.start() + harness = _Harness(scheduler, port) + harness.start() + try: + # The SECOND attempt (after the deadline fired) completes the handshake. + conn = scheduler.wait_connection(0) + assert conn.hello is not None + assert stalls["n"] >= 2 + finally: + harness.stop() + server.stop(grace=0) + + +def test_not_leader_redirect_is_followed() -> None: + """FAILED_PRECONDITION not_leader: redirects the worker to the + leader immediately; schemeless targets keep the dialing TLS mode (#372, + plaintext here on both ends).""" + real = FakeScheduler() + real_server = grpc.server(futures.ThreadPoolExecutor(max_workers=8)) + pb_grpc.add_WorkerSchedulerServicer_to_server(real, real_server) + real_port = real_server.add_insecure_port("127.0.0.1:0") + real_server.start() + + class _NotLeader(FakeScheduler): + def Connect(self, request_iterator, context): + context.abort(grpc.StatusCode.FAILED_PRECONDITION, + f"not_leader:127.0.0.1:{real_port}") + + stale = _NotLeader() + stale_server = grpc.server(futures.ThreadPoolExecutor(max_workers=4)) + pb_grpc.add_WorkerSchedulerServicer_to_server(stale, stale_server) + stale_port = stale_server.add_insecure_port("127.0.0.1:0") + stale_server.start() + + harness = _Harness(stale, stale_port) + harness.start() + try: + conn = real.wait_connection(0) + assert conn.hello is not None and conn.hello.worker_id == "e2e-worker" + finally: + harness.stop() + real_server.stop(grace=0) + stale_server.stop(grace=0) + + +def test_normalize_grpc_addr_inherits_tls_for_schemeless_redirects() -> None: + from gen_worker.transport import normalize_grpc_addr + + assert normalize_grpc_addr("10.0.0.1:7000", default_tls=True) == ("10.0.0.1:7000", True) + assert normalize_grpc_addr("10.0.0.1:7000", default_tls=False) == ("10.0.0.1:7000", False) + # Explicit schemes always win. + assert normalize_grpc_addr("grpc://10.0.0.1:7000", default_tls=True) == ("10.0.0.1:7000", False) + assert normalize_grpc_addr("grpcs://10.0.0.1:7000", default_tls=False) == ("10.0.0.1:7000", True) + # No hint: the bare :443 heuristic stands. + assert normalize_grpc_addr("example.com:443") == ("example.com:443", True) + assert normalize_grpc_addr("example.com:7000") == ("example.com:7000", False) + + # --------------------------------------------------------------------------- # Send-queue policy (#357): results are never dropped; progress sheds oldest. # --------------------------------------------------------------------------- diff --git a/uv.lock b/uv.lock index 6bc308d..35bcdd4 100644 --- a/uv.lock +++ b/uv.lock @@ -55,15 +55,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/da/42/e921fccf5015463e32a3cf6ee7f980a6ed0f395ceeaa45060b61d86486c2/anyio-4.13.0-py3-none-any.whl", hash = "sha256:08b310f9e24a9594186fd75b4f73f4a4152069e3853f1ed8bfbf58369f4ad708", size = 114353, upload-time = "2026-03-24T12:59:08.246Z" }, ] -[[package]] -name = "backoff" -version = "2.2.1" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/47/d7/5bbeb12c44d7c4f2fb5b56abce497eb5ed9f34d85701de869acedd602619/backoff-2.2.1.tar.gz", hash = "sha256:03f829f5bb1923180821643f8753b0502c3b682293992485b0eef2807afa5cba", size = 17001, upload-time = "2022-10-05T19:19:32.061Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/df/73/b6e24bd22e6720ca8ee9a85a0c4a2971af8497d8f3193fa05390cbd46e09/backoff-2.2.1-py3-none-any.whl", hash = "sha256:63579f9a0628e06278f7e47b7d7d5b6ce20dc65c5e96a6f3ca99a6adca0396e8", size = 15148, upload-time = "2022-10-05T19:19:30.546Z" }, -] - [[package]] name = "bitsandbytes" version = "0.49.2" @@ -609,7 +600,6 @@ name = "gen-worker" version = "0.8.3" source = { editable = "." } dependencies = [ - { name = "backoff" }, { name = "blake3" }, { name = "boto3" }, { name = "grpcio" }, @@ -654,7 +644,6 @@ vision = [ [package.metadata] requires-dist = [ - { name = "backoff", specifier = ">=2.2.1" }, { name = "blake3", specifier = ">=1.0.0" }, { name = "boto3", specifier = ">=1.41.0" }, { name = "grpcio", specifier = ">=1.80.0" },