uploader: default to eight workers and harden I/O

Author: tonyredondo

docs/Configuration_Reference.md

@@ -303,7 +303,7 @@ Rule: `dd_payload_uploader(...)`
 | `keep_payloads` | bool | `False` | Keep payload files after successful upload |
 | `filter_prefix` | bool | `False` | Only upload files matching `span_events_*.json` or `coverage_*.json` |
 | `gzip_payloads` | bool | `False` | Gzip test payloads before upload |
-| `workers` | int | `4` | Maximum independent payload-file workers in Python mode; `DD_TEST_OPTIMIZATION_WORKERS` and `--workers` override it at runtime |
+| `workers` | int | `8` | Maximum independent payload-file workers in Python mode; `DD_TEST_OPTIMIZATION_WORKERS` and `--workers` override it at runtime |
 | `use_python_uploader` | bool | `False` | Opt in to the parallel Python uploader; requires host Python 3.10+ at uploader runtime |
 | `data` | label_list | `[]` | Data files to include (for example, `context.json` for enrichment) |
 | `expected_targets` | string_list | `[]` | Optional exact local labels expected in the matching BEP. Fresh and cached results jointly satisfy coverage; only fresh outputs are inspected or uploaded |

docs/Uploader_Reference.md

@@ -172,7 +172,7 @@ dd_test_optimization_targets(
     sync_repo_name = "test_optimization_data",
     uploader_kwargs = {
         "use_python_uploader": True,
-        "workers": 4,
+        "workers": 8,
     },
 )
 ```
@@ -186,7 +186,7 @@ worker can process test, coverage, or telemetry payloads; chunks and other
 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`
+rule-level `workers` value defaults to `8`; `DD_TEST_OPTIMIZATION_WORKERS`
 overrides it at runtime and `--workers=<positive-integer>` has highest
 precedence. Leave
 `use_python_uploader = False` (the default) to retain the legacy Bash or
@@ -301,7 +301,7 @@ bazel run --config=test-optimization //:dd_upload_payloads
 | `DD_TEST_OPTIMIZATION_FILTER_PREFIX` | `0` | `0` uploads all payload files; set to `1` to only upload `span_events_*.json` or `coverage_*.json` |
 | `DD_TEST_OPTIMIZATION_DEBUG` | `0` | Set to `1` to enable verbose upload logging (HTTP codes, response bodies, startTime stats, and key runfile/CODEOWNERS resolution hits) |
 | `DD_TEST_OPTIMIZATION_GZIP` | `0` | Set to `1` to gzip test payloads before upload (adds `Content-Encoding: gzip`) |
-| `DD_TEST_OPTIMIZATION_WORKERS` | rule `workers` (`4`) | Override the maximum independent payload-file workers in Python mode; `--workers` has higher precedence |
+| `DD_TEST_OPTIMIZATION_WORKERS` | rule `workers` (`8`) | Override the maximum independent payload-file workers in Python mode; `--workers` has higher precedence |
 | `DD_TEST_OPTIMIZATION_MAX_WAIT_SEC` | `300` | Override max wait time for slow filesystems (NFS, network drives); set to `0` to skip waiting when no payloads are present |
 | `DD_TEST_OPTIMIZATION_QUIESCENT_SEC` | `10` | Override quiescence wait time |
 | `DD_TEST_OPTIMIZATION_MAX_DEPTH` | `0` (unlimited) | Limit `find` depth for large `bazel-testlogs` trees |

docs/python_parallel_uploader_migration_plan.md

@@ -577,9 +577,17 @@ Measure with identical payload fixtures and backend behavior:
 - temporary bytes written;
 - p50/p95 per-file latency.
 
-Compare `workers=1`, the legacy uploader, and at least `workers=2/4/8`. The
-worker default remains conservative (`4`) until supported CI hosts show a
-repeatable benefit without backend throttling or excessive memory/temp usage.
+Compare `workers=1`, the legacy uploader, and at least `workers=2/4/8`.
+
+A local ARM64/Python 3.12 loopback benchmark at commit `123113e` measured the
+real worker pipeline with identical fixtures and rotating execution order. For
+48 small mixed test/coverage/telemetry files, median speedups over one worker
+were `1.99x`, `3.83x`, and `6.14x` with 2, 4, and 8 workers. For eight test
+payloads above the 4.5 MiB split threshold, the corresponding speedups were
+`1.40x`, `1.50x`, and `1.51x`. Based on the network-bound large-batch result,
+the default is `8`; consumers with split-heavy workloads or backend throttling
+can override it to `4`. Continue validating memory, temporary storage, and
+real-backend retry behavior on supported CI hosts.
 
 ## Rollout and Rollback
 

tools/core/test_optimization_uploader.bzl

@@ -977,7 +977,7 @@ _dd_payload_uploader_rule = rule(
         "keep_payloads": attr.bool(default = False, doc = "Keep payload files after successful upload (env: DD_TEST_OPTIMIZATION_KEEP_PAYLOADS)"),
         "filter_prefix": attr.bool(default = False, doc = "Boolean gate: only upload files matching span_events_*.json or coverage_*.json; telemetry uploads are always eligible (env: DD_TEST_OPTIMIZATION_FILTER_PREFIX)"),
         "gzip_payloads": attr.bool(default = False, doc = "Gzip test payloads before upload (env: DD_TEST_OPTIMIZATION_GZIP)"),
-        "workers": attr.int(default = 4, doc = "Maximum number of independent payload-file workers (env: DD_TEST_OPTIMIZATION_WORKERS, CLI: --workers)"),
+        "workers": attr.int(default = 8, doc = "Maximum number of independent payload-file workers (env: DD_TEST_OPTIMIZATION_WORKERS, CLI: --workers)"),
         "use_python_uploader": attr.bool(default = False, doc = "Temporary rollout switch selecting the Python uploader implementation."),
         # Optional files to place in runfiles (e.g., a generated context.json)
         "data": attr.label_list(allow_files = True, doc = "Data files to include in runfiles (e.g., context.json for enrichment)"),

tools/core/uploader_py/discovery.py

@@ -138,6 +138,11 @@ def discover_file_tasks(
         for payload_type, relative_directory in _PAYLOAD_SUBDIRECTORIES:
             payload_directory = output.path / relative_directory
             if not payload_directory.is_dir():
+                if payload_directory.is_symlink():
+                    warnings.append("payload_symlink_skipped")
+                continue
+            if not _safe_payload_directory(payload_directory, output.path):
+                warnings.append("payload_symlink_skipped")
                 continue
             for source_path in sorted(payload_directory.iterdir(), key=lambda item: item.name):
                 if source_path.is_symlink():
@@ -267,3 +272,17 @@ def _find_test_outputs(root: Path, *, max_depth: int) -> tuple[Path, ...]:
         if max_depth > 0 and depth >= max_depth:
             subdirectories.clear()
     return tuple(sorted(found, key=lambda path: path.as_posix()))
+
+
+def _safe_payload_directory(path: Path, output: Path) -> bool:
+    """Accept only real payload directories contained by their test output."""
+    current = output
+    for part in path.relative_to(output).parts:
+        current /= part
+        if current.is_symlink():
+            return False
+    try:
+        path.resolve().relative_to(output.resolve())
+    except (OSError, ValueError):
+        return False
+    return True

tools/core/uploader_py/freshness.py

@@ -500,11 +500,7 @@ def _build_plan(
 
 def _parse_execution_log(path: Path) -> set[tuple[str, str]]:
     eligible: set[tuple[str, str]] = set()
-    try:
-        lines = path.read_text(encoding="utf-8-sig").splitlines()
-    except OSError as exc:
-        raise FreshnessError(f"failed to read execution log JSON {path}: {exc}") from exc
-    for line_number, raw_line in enumerate(lines, start=1):
+    for line_number, raw_line in enumerate(_stream_execution_log(path), start=1):
         if not raw_line.strip():
             continue
         try:
@@ -540,6 +536,15 @@ def _parse_execution_log(path: Path) -> set[tuple[str, str]]:
     return eligible
 
 
+def _stream_execution_log(path: Path) -> Iterable[str]:
+    """Yield one action record at a time so large logs stay memory-bounded."""
+    try:
+        with path.open("r", encoding="utf-8-sig") as lines:
+            yield from lines
+    except (OSError, UnicodeError) as exc:
+        raise FreshnessError(f"failed to read execution log JSON {path}: {exc}") from exc
+
+
 def _execution_output_key(raw: str) -> str:
     normalized = raw.replace("\\", "/")
     if "/testlogs/" in normalized:

tools/core/uploader_py/models.py

@@ -17,7 +17,7 @@
 
 
 MAX_TEST_PAYLOAD_BYTES = 4_718_592
-DEFAULT_WORKERS = 4
+DEFAULT_WORKERS = 8
 
 
 class PayloadType(str, Enum):

tools/core/uploader_py/transport.py

@@ -609,9 +609,14 @@ def _post(
                     headers=dict(prepared_request.headers),
                     method="POST",
                 )
-                with closing(
-                    self._opener.open(http_request, timeout=self.request_timeout)
-                ) as response:
+                try:
+                    response = self._opener.open(
+                        http_request,
+                        timeout=self.request_timeout,
+                    )
+                except HTTPError as exc:
+                    response = exc
+                with closing(response):
                     status_code = int(response.status)
                     response_excerpt, excerpt_truncated = _bounded_response(
                         response,
@@ -633,17 +638,13 @@ def _post(
                         )
                     retryable = _retryable_status(status_code)
                     retry_after = response.headers.get("Retry-After")
-            except HTTPError as exc:
-                with closing(exc):
-                    status_code = int(exc.code)
-                    response_excerpt, excerpt_truncated = _bounded_response(
-                        exc,
-                        self.response_limit,
-                    )
-                    retry_after = exc.headers.get("Retry-After")
-                transport_error = None
-                retryable = _retryable_status(status_code)
-            except (URLError, TimeoutError, ConnectionError, http.client.HTTPException) as exc:
+            except (
+                URLError,
+                TimeoutError,
+                ConnectionError,
+                http.client.HTTPException,
+                ssl.SSLError,
+            ) as exc:
                 status_code = None
                 response_excerpt = b""
                 excerpt_truncated = False

tools/tests/python/test_uploader_discovery_tools.py

@@ -117,6 +117,43 @@ def test_payload_symlink_is_never_scheduled(self) -> None:
             self.assertEqual((), discovery.tasks)
             self.assertIn("payload_symlink_skipped", discovery.warning_codes)
 
+    def test_payload_directory_symlink_is_never_scheduled(self) -> None:
+        with tempfile.TemporaryDirectory() as raw_root:
+            root = Path(raw_root)
+            output = root / "pkg" / "target" / "test.outputs"
+            outside = root / "outside" / "tests"
+            outside.mkdir(parents=True)
+            (outside / "events.json").write_text("{}", encoding="utf-8")
+            link = output / "payloads" / "tests"
+            link.parent.mkdir(parents=True)
+            try:
+                link.symlink_to(outside, target_is_directory=True)
+            except OSError as exc:
+                self.skipTest(f"directory symlink creation unavailable: {exc}")
+
+            discovery = discover_file_tasks((ScanRoot(root),))
+
+            self.assertEqual((), discovery.tasks)
+            self.assertIn("payload_symlink_skipped", discovery.warning_codes)
+
+    def test_intermediate_payload_symlink_cannot_escape_test_outputs(self) -> None:
+        with tempfile.TemporaryDirectory() as raw_root:
+            root = Path(raw_root)
+            output = root / "pkg" / "target" / "test.outputs"
+            outside = root / "outside" / "payloads"
+            (outside / "tests").mkdir(parents=True)
+            (outside / "tests" / "events.json").write_text("{}", encoding="utf-8")
+            output.mkdir(parents=True)
+            try:
+                (output / "payloads").symlink_to(outside, target_is_directory=True)
+            except OSError as exc:
+                self.skipTest(f"directory symlink creation unavailable: {exc}")
+
+            discovery = discover_file_tasks((ScanRoot(root),))
+
+            self.assertEqual((), discovery.tasks)
+            self.assertIn("payload_symlink_skipped", discovery.warning_codes)
+
     def test_selected_output_never_falls_back_to_stale_local_when_staging_failed(self) -> None:
         with tempfile.TemporaryDirectory() as raw_root:
             root = Path(raw_root)

tools/tests/python/test_uploader_freshness_tools.py

@@ -16,6 +16,7 @@
 from pathlib import Path
 import tempfile
 import unittest
+from unittest import mock
 import zipfile
 
 from uploader_test_support import (
@@ -135,11 +136,16 @@ def test_execution_log_selects_only_fresh_output_and_stamps_label(self) -> None:
             )
             resolver = RunfilesResolver.from_environment(cwd=root, environ={})
 
-            prepared = prepare_freshness(
-                config,
-                resolver=resolver,
-                local_testlogs_root=testlogs,
-            )
+            with mock.patch.object(
+                Path,
+                "read_text",
+                side_effect=AssertionError("execution logs must be streamed"),
+            ):
+                prepared = prepare_freshness(
+                    config,
+                    resolver=resolver,
+                    local_testlogs_root=testlogs,
+                )
             discovery = discover_file_tasks(prepared.scan_roots)
             filtered = filter_discovery_for_freshness(
                 discovery,