uploader: harden upload preflight and retry handling

Author: tonyredondo

docs/Uploader_Reference.md

@@ -183,8 +183,9 @@ This mode requires a host Python 3.10 or newer on Linux, macOS, and Windows at
 performs its enrichment, validation, preflight split at the conservative
 `4_718_592`-byte (4.5 MiB) threshold, and upload/retries independently. Every
 worker can process test, coverage, or telemetry payloads; chunks and other
-derived requests belonging to one source remain sequential. Non-standard JSON
-numbers (`NaN` and positive/negative `Infinity`) are rejected before HTTP. The
+derived requests belonging to one source remain sequential. Test and telemetry
+JSON reject non-standard numbers (`NaN` and positive/negative `Infinity`) before
+HTTP; coverage bodies remain opaque JSON/msgpack multipart parts. The
 rule-level `workers` value defaults to `4`; `DD_TEST_OPTIMIZATION_WORKERS`
 overrides it at runtime and `--workers=<positive-integer>` has highest
 precedence. Leave
@@ -691,10 +692,14 @@ payload discovery/quiescence before proceeding.
 - Each logical request makes at most four total attempts: the initial attempt
   plus up to three retries, normally separated by 2 seconds.
 - Connection failures, timeouts, HTTP `408`, HTTP `429`, and HTTP `5xx` are
-  retryable. `Retry-After` is honored when the backend supplies it.
-- Other HTTP `4xx` responses are terminal after the first attempt. In
-  particular, `413` indicates that the preventive split contract was violated;
-  it is never retried and never triggers adaptive splitting.
+  retryable. `Retry-After` is honored when the backend supplies it, with a
+  60-second safety cap per retry so a response cannot stall a worker
+  indefinitely.
+- Other HTTP `4xx` responses are terminal after the first attempt. For test
+  payloads, `413` indicates that the preventive split contract was violated.
+  Coverage and telemetry are not split, so their `413` failures use the
+  `upload_http_413` reason instead. No `413` is retried or triggers adaptive
+  splitting.
 - JSON, gzip, telemetry, and multipart bodies are prepared once per logical
   request and replayed byte-for-byte for every retry.
 - The temporary legacy Bash/curl and PowerShell implementations remain

docs/python_parallel_uploader_migration_plan.md

@@ -46,11 +46,12 @@ Current checkpoint:
   schema can be shared by workers while each validation keeps independent
   errors, warnings, and counters; the existing validator CLI remains intact;
 - a standard-library HTTP transport now implements the normalized
-  four-attempt retry policy, terminal `413`/permanent `4xx`, `Retry-After`,
-  separate connect and socket-I/O timeouts, bounded response diagnostics,
-  system TLS verification, snapshotted proxy/`NO_PROXY` settings, gzip, and
-  replayable exact-length JSON/multipart bodies. Its local macOS compatibility
-  spike passes; HTTPS handshakes and Linux/Windows lanes remain open gates;
+  four-attempt retry policy, terminal `413`/permanent `4xx`, `Retry-After`
+  bounded to 60 seconds per retry, separate connect and socket-I/O timeouts,
+  bounded response diagnostics, system TLS verification, snapshotted
+  proxy/`NO_PROXY` settings, gzip, and replayable exact-length JSON/multipart
+  bodies. Its local macOS compatibility spike passes; HTTPS handshakes and
+  Linux/Windows lanes remain open gates;
 - CODEOWNERS discovery and parsing now produce one immutable matcher before
   worker startup. It preserves current discovery order, last-match-wins,
   explicit empty-owner rules, source candidate normalization, producer-owned
@@ -81,8 +82,10 @@ Current checkpoint:
   enrichment-tag checks, preventive splitting, gzip preparation, telemetry
   augmentation, and the upload path's URL/header/body-size request validation,
   then stops before HTTP and source cleanup;
-- custom agentless and EVP base URLs now validate numeric in-range ports during
-  preflight, so dry-run rejects an endpoint that upload mode could not open;
+- custom agentless and EVP base URLs now validate numeric in-range ports,
+  printable ASCII, and percent escapes during preflight. Effective HTTP(S)
+  proxy configuration is also validated before workers start, so dry-run
+  rejects request configuration that upload mode could not use;
 - every JSON document parsed or generated by the Python uploader now rejects
   non-standard `NaN` and positive/negative `Infinity` values before HTTP;
 - the deterministic compact-JSON splitter is connected to the worker path and
@@ -134,7 +137,7 @@ Current checkpoint:
 Validation checkpoint (`2026-08-29`, macOS arm64):
 
 - `python3 -m unittest discover -s tools/tests/python -p 'test*_tools.py'`:
-  446 tests executed; 445 pass and one platform-specific test is skipped on
+  450 tests executed; 449 pass and one platform-specific test is skipped on
   macOS with ShellCheck and PowerShell installed;
 - `./bazelw test //tools/tests/core:tests
   //tools/tests/python:python_tools_test
@@ -751,7 +754,8 @@ dry-run modes.
 | Optional telemetry provider/facts augmentation cannot be produced | Warning | Send the valid primary body when possible; omit only the unavailable optional derived request | Retain current best-effort augmentation behavior |
 | Request construction fails | File failure | No request, or stop remaining derived requests | Retain source |
 | Retry budget is exhausted or a terminal HTTP status is received | File failure | Stop later chunks/derived requests for that source | Retain source; unrelated workers continue |
-| HTTP `413` | Immediate file failure | Do not retry and do not split again | Retain source with `payload_limit_contract_mismatch` |
+| Test HTTP `413` after preventive splitting | Immediate file failure | Do not retry and do not split again | Retain source with `payload_limit_contract_mismatch` |
+| Coverage or telemetry HTTP `413` | Immediate file failure | Do not retry; structural splitting is unsupported for these types | Retain source with `upload_http_413` |
 | Source deletion fails after successful delivery | Warning | Do not repeat the successful request | Report the source as retained without converting intake success into upload failure |
 
 Warnings must be captured in debug/report diagnostics without incrementing
@@ -1043,7 +1047,7 @@ Initial policy:
 - 60-second request timeout;
 - retry transport failures that can be transient;
 - retry HTTP `408`, `429`, and `5xx` statuses;
-- honor a valid `Retry-After` header;
+- honor a valid `Retry-After` header up to a 60-second safety cap per retry;
 - otherwise use the documented two-second retry delay for initial parity;
 - do not retry permanent `4xx` responses;
 - specifically do not retry `413`;
@@ -1741,6 +1745,8 @@ validation describe and exercise the same implementation.
 - Timeout then success.
 - `408` retry.
 - `429` with `Retry-After` retry.
+- Normal, excessive numeric, far-future date, and overflow-sized `Retry-After`
+  values remain finite and do not exceed the 60-second cap.
 - `500`, `502`, `503`, and `504` retry.
 - Sustained retryable failure exhausts the budget.
 - Permanent `400`, `401`, `403`, and `404` are not retried.

tools/core/uploader_py/application.py

@@ -52,6 +52,7 @@
 from .reporting import AggregateReport, LegacyReportContext
 from .resources import ResourceInputs, load_resources
 from .temporary import TemporaryDirectoryError
+from .transport import HttpTransportError
 from .worker_pool import WorkerPoolError
 
 
@@ -61,6 +62,7 @@
     ExpectedTargetsError,
     FreshnessError,
     TemporaryDirectoryError,
+    HttpTransportError,
     WorkerPoolError,
     WorkspaceLockError,
 )

tools/core/uploader_py/coordinator.py

@@ -31,7 +31,7 @@
 from .resources import LoadedResources
 from .telemetry import build_telemetry_plan
 from .temporary import invocation_temporary_directory
-from .transport import HttpTransport
+from .transport import HttpTransport, validate_proxy_environment
 from .worker_pool import (
     WorkerPoolInterrupted,
     WorkerPoolRun,
@@ -97,6 +97,7 @@ def execute_discovery(
 ) -> CoordinatorOutcome:
     """Run already-authorized tasks; workers never mutate coordinator state."""
     started = clock()
+    validate_proxy_environment(settings.proxy_environment)
     matcher = load_codeowners_matcher(
         explicit_path=settings.codeowners_file,
         workspace_root=settings.workspace,

tools/core/uploader_py/endpoints.py

@@ -10,14 +10,15 @@
 
 from dataclasses import dataclass
 import re
-from urllib.parse import urlsplit
+from urllib.parse import SplitResult, urlsplit
 
 from .config import ConfigError, UploaderConfig
 
 
 _VALID_HOSTNAME_RE = re.compile(
     r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?([.][a-z0-9]([a-z0-9-]*[a-z0-9])?)*$"
 )
+_INVALID_PERCENT_ESCAPE_RE = re.compile(r"%(?![0-9A-Fa-f]{2})")
 
 
 @dataclass(frozen=True)
@@ -60,32 +61,41 @@ def normalize_dd_site(raw_site: str) -> str:
     return site
 
 
+def parse_http_url(raw_url: str) -> SplitResult:
+    """Parse one network-ready absolute HTTP(S) URL.
+
+    urllib accepts some values during parsing that ``http.client`` rejects only
+    when it starts a request. Keeping this small preflight shared by endpoint
+    configuration and request preparation makes dry-run reject those values too.
+    """
+    if not isinstance(raw_url, str) or not raw_url:
+        raise ValueError("HTTP URL must be a non-empty string")
+    if any(ord(character) < 33 or ord(character) > 126 for character in raw_url):
+        raise ValueError("HTTP URL must contain only printable ASCII characters")
+    if _INVALID_PERCENT_ESCAPE_RE.search(raw_url):
+        raise ValueError("HTTP URL contains an invalid percent escape")
+
+    parsed = urlsplit(raw_url)
+    hostname = parsed.hostname or ""
+    # Accessing port performs urllib's numeric and range validation.
+    _ = parsed.port
+    if parsed.scheme.lower() not in {"http", "https"} or not hostname:
+        raise ValueError("HTTP URL must be absolute")
+    return parsed
+
+
 def _validated_base_url(raw_url: str, variable_name: str) -> str:
     """Validate an endpoint override without echoing sensitive URL components."""
     base = raw_url.rstrip("/")
     try:
-        parsed = urlsplit(base)
-        hostname = parsed.hostname or ""
-        # Accessing port performs urllib's numeric and range validation.
-        _ = parsed.port
-        valid = (
-            parsed.scheme.lower() in {"http", "https"}
-            and bool(hostname)
-            and not any(
-                ord(character) <= 32 or ord(character) == 127
-                for character in hostname
-            )
-        )
+        parsed = parse_http_url(base)
     except (TypeError, ValueError):
-        valid = False
         parsed = None
-    if not valid:
+    if parsed is None:
         raise ConfigError(f"{variable_name} must be an absolute HTTP(S) URL")
-    if parsed is not None and (
-        parsed.username is not None or parsed.password is not None
-    ):
+    if parsed.username is not None or parsed.password is not None:
         raise ConfigError(f"{variable_name} must not contain credentials/userinfo")
-    if parsed is not None and (parsed.query or parsed.fragment):
+    if parsed.query or parsed.fragment:
         raise ConfigError(f"{variable_name} must not contain a query or fragment")
     return base
 

tools/core/uploader_py/file_worker.py

@@ -467,6 +467,7 @@ def _process_test(
         if not result.succeeded:
             failure_code, failure_message = _http_failure(
                 result,
+                payload_type=task.payload_type,
                 payload_limit_context=(
                     f"chunk={chunk.index}/{len(chunks)} "
                     f"uncompressed_bytes={chunk.size_bytes} "
@@ -583,7 +584,10 @@ def _process_coverage(
         f"coverage request completed after {result.attempts} attempt(s)",
     )
     if not result.succeeded:
-        failure_code, failure_message = _http_failure(result)
+        failure_code, failure_message = _http_failure(
+            result,
+            payload_type=task.payload_type,
+        )
         return FileResult(
             status=FileStatus.FAILED,
             requests_attempted=result.attempts,
@@ -718,7 +722,10 @@ def _process_telemetry(
             f"telemetry request completed after {result.attempts} attempt(s)",
         )
         if not result.succeeded:
-            failure_code, failure_message = _http_failure(result)
+            failure_code, failure_message = _http_failure(
+                result,
+                payload_type=task.payload_type,
+            )
             return FileResult(
                 status=FileStatus.FAILED,
                 requests_attempted=requests_attempted,
@@ -995,15 +1002,20 @@ def _cleanup_source(path: Path, keep_payloads: bool) -> tuple[bool, str | None]:
 def _http_failure(
     result: HttpResult,
     *,
+    payload_type: PayloadType,
     payload_limit_context: str | None = None,
 ) -> tuple[str, str]:
     if result.status_code == 413:
-        if payload_limit_context:
+        if payload_type is PayloadType.TEST and payload_limit_context:
             return (
                 "payload_limit_contract_mismatch",
                 f"HTTP 413 after preventive split; {payload_limit_context}",
             )
-        return "upload_http_413", "HTTP 413 after preventive split"
+        return (
+            "upload_http_413",
+            f"HTTP 413 for unsplit {payload_type.value} payload; "
+            f"{payload_type.value} splitting is not supported",
+        )
     if result.status_code is not None:
         return "upload_http_error", f"HTTP {result.status_code}"
     return "upload_transport_error", result.transport_error or "transport error"