chore(deps): update Ray to 2.57.0

docker/Dockerfile

@@ -68,11 +68,6 @@ COPY docker/common/install_etcd_nats.sh .
 RUN bash install_etcd_nats.sh && \
     rm install_etcd_nats.sh
 
-# Use the ray-haproxy wheel so Ray Serve's HAProxy ingress mode can discover
-# the packaged binary without compiling HAProxy in the image.
-ENV RAY_SERVE_EXPERIMENTAL_PIP_HAPROXY=1
-
-
 FROM nemo_curator_dep AS nemo_curator
 
 ARG CURATOR_EXTRA=all
@@ -90,13 +85,6 @@ RUN if [ -n "${CURATOR_EXTRA}" ]; then \
         uv sync --link-mode copy --locked ${CURATOR_GROUPS} --no-cache; \
     fi && \
     find /opt/venv -type d -path "*ray/_private/runtime_env/agent/thirdparty_files/aiohttp*" -exec rm -rf {} + && \
-    # Address GHSA-72hv-8253-57qq: jackson-core bundled in ray_dist.jar (jackson-core 2.16.1 < 2.18.6)
-    # Ray's Java support is not used; deleting the JAR removes the vulnerable jackson-core classes.
-    # TODO: Remove once Ray upgrades bundled jackson-core to >=2.18.6 (https://github.com/ray-project/ray/issues/35949)
-    find /opt/venv -name "ray_dist.jar" -delete && \
-    if find /opt/venv -name "ray_dist.jar" | grep -q .; then \
-        echo "ERROR: GHSA-72hv-8253-57qq not fixed — ray_dist.jar still present after deletion" && exit 1; \
-    fi && \
     # Strip opencv-python-headless (ffmpeg CVE carrier); tolerate arm64 where vllm doesn't pull it.
     (uv pip uninstall opencv-python-headless || true)
 

fern/versions/main/pages/curate-text/synthetic/inference-server.mdx

@@ -52,9 +52,9 @@ The stack uses:
 
 - CUDA 12.9.1 in the NeMo Curator container.
 - PyTorch 2.11.
-- Ray Serve 2.56.1 or later.
+- Ray Serve 2.57.0 or later.
 - vLLM 0.22.0 with CUDA 12.9.
-- A public ai-dynamo release 1.3.0 or later and NIXL 0.10.0 or later for the
+- A public ai-dynamo release 1.3.1 or later and NIXL 0.10.0 or later for the
   Dynamo backend.
 
 Start or connect to a Ray cluster before creating the server. The examples below use `RayClient`; an externally managed Ray cluster works as well.
@@ -375,9 +375,9 @@ Dynamo creates detached, named placement groups and actors. On startup, it remov
 
 ## HAProxy Ingress
 
-The NeMo Curator container includes HAProxy and `socat`. When both binaries are available, local Ray cluster initialization enables Ray Serve's HAProxy ingress and assigns a free metrics port. If either binary is unavailable, Ray Serve uses its default Python proxy.
+Ray Serve's Linux dependencies include the bundled `ray-haproxy` binary. Local Ray cluster initialization enables HAProxy ingress when that package or an explicit `RAY_SERVE_HAPROXY_BINARY_PATH` is available and assigns free metrics and statistics ports. Otherwise, Ray Serve uses its default Python proxy.
 
-No HAProxy configuration is required in `InferenceServer`. To confirm the optimized path, check startup logs for `Ray Serve HAProxy ingress enabled`.
+No additional proxy package installation or HAProxy configuration is required for `InferenceServer`, which enables HAProxy when starting Ray Serve. To confirm that a local cluster discovered the packaged binary, check startup logs for `Ray Serve HAProxy ingress enabled`.
 
 ## Use with NeMo Curator Clients
 

fern/versions/main/pages/reference/infrastructure/container-environments.mdx

@@ -49,8 +49,8 @@ The primary container includes comprehensive support for all curation modalities
 
 **Pre-installed components:**
 - NeMo Curator with all optional dependencies (`[all]` extras)
-- CUDA 12.8.1 with cuDNN
-- Python 3.12 with uv package manager
+- CUDA 12.9.1 with cuDNN
+- Python 3.13 with uv package manager
 - FFmpeg 8+ with NVENC support (for video processing)
 - Ray and its distributed computing libraries
 - NVIDIA optimized Python packages
@@ -63,8 +63,8 @@ Ray is the execution framework for all NeMo Curator pipelines. Dask is no longer
 
 | Property | Value |
 | --- | --- |
-| Python Version | 3.12 |
-| CUDA Version | 12.8.1 (configurable) |
+| Python Version | 3.13 |
+| CUDA Version | 12.9.1 (configurable) |
 | Operating System | Ubuntu 24.04 (configurable) |
 | Base Image | `nvidia/cuda:${CUDA_VER}-cudnn-devel-${LINUX_VER}` |
 | Package Manager | uv (Ultrafast Python package installer) |
@@ -73,21 +73,13 @@ Ray is the execution framework for all NeMo Curator pipelines. Dask is no longer
 
 ---
 
-## Security Hardening
-
-The container build includes the following security measures:
-
-- **`ray_dist.jar` removal**: Ray's Java support JAR is deleted during the build to remove a bundled jackson-core library affected by [GHSA-72hv-8253-57qq](https://github.com/advisories/GHSA-72hv-8253-57qq) (DoS via async JSON parser). NeMo Curator does not use Ray's Java support, so this has no functional impact. A build-time verification guard fails the build if the JAR is not successfully removed.
-
----
-
 ## Container Build Arguments
 
 The main container accepts these build-time arguments for environment customization:
 
 | Argument | Default | Description |
 |----------|---------|-------------|
-| `CUDA_VER` | `12.8.1` | CUDA version |
+| `CUDA_VER` | `12.9.1` | CUDA version |
 | `LINUX_VER` | `ubuntu24.04` | Base OS version |
 | `CURATOR_ENV` | `ci` | Curator environment type |
 | `NVIDIA_BUILD_ID` | `<unknown>` | NVIDIA build identifier |

nemo_curator/backends/ray_data/diagnostics.py

@@ -30,6 +30,7 @@
 import logging
 import os
 import threading
+import time
 from dataclasses import dataclass
 from enum import StrEnum
 from typing import Any
@@ -38,7 +39,7 @@
 # private, untyped implementation modules.
 # ruff: noqa: ANN401
 
-_SUPPORTED_RAY_VERSION = "2.56.1"
+_SUPPORTED_RAY_VERSION = "2.57.0"
 _INSTALL_MARKER = "_nemo_curator_ray_data_diagnostics_installed"
 _INSTALL_LOCK = threading.Lock()
 RAY_DATA_DIAGNOSTICS_ENV_VAR = "NEMO_CURATOR_RAY_DATA_DIAGNOSTICS"
@@ -66,6 +67,10 @@ class _TaskAdmissionDecision:
     allocation: Any = None
 
 
+def _milliseconds(seconds: float) -> float:
+    return round(seconds * 1000, 3)
+
+
 def format_logfmt_event(event: str, fields: dict[str, object]) -> str:
     """Format an event as stable, parseable logfmt-like tokens."""
 
@@ -94,6 +99,13 @@ def execution_resource_fields(prefix: str, resources: Any) -> dict[str, object]:
     }
 
 
+def _object_store_memory_fields(resource_manager: Any, op: Any) -> dict[str, object]:
+    return {
+        "object_store_internal_bytes": resource_manager.get_mem_op_internal(op),
+        "object_store_output_bytes": resource_manager.get_mem_op_outputs(op),
+    }
+
+
 def install_ray_data_diagnostics() -> DiagnosticsInstallStatus:
     """Install driver-side diagnostics without modifying the Ray installation.
 
@@ -154,7 +166,7 @@ def _has_native_diagnostics(autoscaler_module: Any, resource_manager_module: Any
     )
 
 
-def _install_resource_admission_diagnostics(  # noqa: C901
+def _install_resource_admission_diagnostics(  # noqa: C901, PLR0915
     resource_manager_module: Any, resource_policy_module: Any
 ) -> None:
     allocator_cls = resource_manager_module.OpResourceAllocator
@@ -201,6 +213,7 @@ def get_reservation_decision(self: Any, op: Any) -> _TaskAdmissionDecision:
     def policy_init(self: Any, data_context: Any, topology: Any, resource_manager: Any) -> None:
         original_init(self, data_context, topology, resource_manager)
         self._nemo_curator_previous_decisions = {}
+        self._nemo_curator_resource_blocked_since = {}
 
     def can_add_input(self: Any, op: Any) -> bool:
         allocator = self._resource_manager._op_resource_allocator
@@ -213,6 +226,13 @@ def can_add_input(self: Any, op: Any) -> bool:
         signature = (decision.allowed, decision.reason)
         previous = self._nemo_curator_previous_decisions
         if previous.get(op) != signature:
+            blocked_since = self._nemo_curator_resource_blocked_since
+            if decision.allowed:
+                started_at = blocked_since.pop(op, None)
+                blocked_duration_ms = None if started_at is None else _milliseconds(time.perf_counter() - started_at)
+            else:
+                blocked_since.setdefault(op, time.perf_counter())
+                blocked_duration_ms = None
             usage = None
             allocation = None
             if decision.remaining_budget is not None:
@@ -227,6 +247,8 @@ def can_add_input(self: Any, op: Any) -> bool:
                 "pending_output_estimate": decision.pending_output_estimate,
                 **execution_resource_fields("usage", usage),
                 **execution_resource_fields("allocation", allocation),
+                **_object_store_memory_fields(self._resource_manager, op),
+                "blocked_duration_ms": blocked_duration_ms,
             }
             resource_policy_module.logger.debug(format_logfmt_event("ray_data_resource_budget_admission", fields))
             previous[op] = signature
@@ -238,6 +260,11 @@ def can_add_input(self: Any, op: Any) -> bool:
 
 def _install_downstream_capacity_diagnostics(downstream_policy_module: Any) -> None:
     policy_cls = downstream_policy_module.DownstreamCapacityBackpressurePolicy
+    original_init = policy_cls.__init__
+
+    def policy_init(self: Any, *args: Any, **kwargs: Any) -> None:
+        original_init(self, *args, **kwargs)
+        self._nemo_curator_downstream_blocked_since = {}
 
     def should_apply_backpressure(self: Any, op: Any) -> bool:
         if self._should_skip_backpressure(op):
@@ -256,6 +283,13 @@ def should_apply_backpressure(self: Any, op: Any) -> bool:
 
         previous = self._prev_should_backpressure.get(op)
         if previous != result:
+            blocked_since = self._nemo_curator_downstream_blocked_since
+            if result:
+                blocked_since.setdefault(op, time.perf_counter())
+                blocked_duration_ms = None
+            else:
+                started_at = blocked_since.pop(op, None)
+                blocked_duration_ms = None if started_at is None else _milliseconds(time.perf_counter() - started_at)
             queue_bytes = self._get_queue_size_bytes(op)
             downstream_capacity_bytes = self._get_downstream_capacity_size_bytes(op)
             downstream_policy_module.logger.debug(
@@ -269,12 +303,15 @@ def should_apply_backpressure(self: Any, op: Any) -> bool:
                         "queue_ratio": f"{queue_ratio:.2f}",
                         "configured_ratio": self._backpressure_capacity_ratio,
                         "utilized_object_store_budget_fraction": utilized_fraction,
+                        **_object_store_memory_fields(self._resource_manager, op),
+                        "blocked_duration_ms": blocked_duration_ms,
                     },
                 )
             )
             self._prev_should_backpressure[op] = result
         return result
 
+    policy_cls.__init__ = policy_init
     policy_cls._should_apply_backpressure = should_apply_backpressure
 
 
@@ -377,6 +414,7 @@ def log_scaling_decision(  # noqa: PLR0913
                     **execution_resource_fields("allocation", allocation),
                     **execution_resource_fields("usage", usage),
                     **execution_resource_fields("remaining_budget", remaining_budget),
+                    **_object_store_memory_fields(self._resource_manager, op),
                 },
             )
         )

nemo_curator/core/serve/ray_serve/backend.py

@@ -49,13 +49,10 @@ def stop(self) -> None:
             import ray
             from ray import serve
 
-            self._clear_ray_serve_client_cache()
             with ray.init(ignore_reinit_error=True):
                 serve.shutdown()
         except Exception:  # noqa: BLE001
             logger.debug("serve.shutdown() failed (cluster may already be gone)")
-        finally:
-            self._clear_ray_serve_client_cache()
 
         logger.info("Ray Serve stopped")
 
@@ -114,21 +111,6 @@ def _quiet_runtime_env() -> dict[str, Any]:
     def _configure_ray_serve_haproxy() -> None:
         """Set Ray Serve HAProxy defaults before importing Ray Serve."""
         os.environ.setdefault("RAY_SERVE_ENABLE_HA_PROXY", "1")
-        if not os.environ.get("RAY_SERVE_HAPROXY_BINARY_PATH"):
-            os.environ.setdefault("RAY_SERVE_EXPERIMENTAL_PIP_HAPROXY", "1")
-
-    @staticmethod
-    def _clear_ray_serve_client_cache() -> None:
-        """Clear Ray Serve's cached controller client before reconnecting for shutdown.
-
-        TODO: Remove this workaround once https://github.com/ray-project/ray/issues/64647 is fixed.
-        """
-        try:
-            from ray.serve.context import _set_global_client
-        except (ImportError, AttributeError):
-            return
-
-        _set_global_client(None)
 
     @staticmethod
     def _to_llm_config(model: RayServeModelConfig, quiet_runtime_env: dict[str, Any] | None = None) -> "LLMConfig":

nemo_curator/core/utils.py

@@ -201,7 +201,6 @@ def init_cluster(  # noqa: PLR0913
             bind_host="0.0.0.0",  # noqa: S104
         )
         os.environ["RAY_SERVE_ENABLE_HA_PROXY"] = "1"
-        os.environ["RAY_SERVE_EXPERIMENTAL_PIP_HAPROXY"] = "1"
         os.environ["RAY_SERVE_HAPROXY_METRICS_PORT"] = str(haproxy_metrics_port)
         os.environ["RAY_SERVE_HAPROXY_STATS_PORT"] = str(haproxy_stats_port)
         logger.info(

pyproject.toml

@@ -68,7 +68,7 @@ dependencies = [
     "openai>=1.0.0",
     "pandas>=2.1.0",
     "pyarrow",
-    "ray[default,data]>=2.56.1",
+    "ray[default,data]>=2.57.0",
     "torch",
     "transformers",
 ]
@@ -90,10 +90,9 @@ vllm = [
 inference_server = [
     "nemo_curator[cuda12]",
     "nemo_curator[vllm]",
-    "ai-dynamo>=1.3.0; platform_system == 'Linux' and (platform_machine == 'x86_64' or platform_machine == 'aarch64')",
+    "ai-dynamo>=1.3.1; platform_system == 'Linux' and (platform_machine == 'x86_64' or platform_machine == 'aarch64')",
     "nixl-cu12>=0.10.0; platform_system == 'Linux' and (platform_machine == 'x86_64' or platform_machine == 'aarch64')",
-    "ray[serve]>=2.56.1",
-    "ray-haproxy==2.8.20; platform_system == 'Linux' and (platform_machine == 'x86_64' or platform_machine == 'aarch64')",
+    "ray[serve]>=2.57.0",
 ]
 
 # Installs CPU + GPU text curation modules
@@ -341,7 +340,7 @@ constraint-dependencies = [
     "cryptography>=46.0.6", # Address CVE GHSA-m959-cc7f-wv43
     "nbconvert>=7.17.0", # Address CVE GHSA-xm59-rqc7-hhvf
     "pillow>=12.1.1", # Address CVE GHSA-cfh3-3jmp-rvhc
-    "protobuf>=5.29.6,<7.0", # Address CVE GHSA-8qvm-5x2c-j2w7; <7.0 due to ray serve FieldDescriptor API breakage
+    "protobuf>=5.29.6", # Address CVE GHSA-8qvm-5x2c-j2w7
     "pyasn1>=0.6.2", # Address CVE GHSA-63vm-454h-vhhq
     "pyopenssl>=26.0.0", # Address CVE GHSA-5pwr-322w-8jr4
     "python-multipart>=0.0.22", # Address CVE GHSA-wp53-j4wj-2cfg
@@ -360,7 +359,7 @@ override-dependencies = [
     "levenshtein;  sys_platform == 'never'",
     "numpy>=2.0.0,<=2.2.0", # Override nemo-toolkits constraint of <2.0.0, upperbounds for Numba compatibility
     "numba==0.65.0", # Override RAPIDS/legacy caps for the inference image; vLLM 0.22 requires numba 0.65.0
-    "protobuf>=5.29.5,<7.0",  # Override nemo-toolkits constraint of ~=5.29.5; <7.0 due to ray serve FieldDescriptor API breakage
+    "protobuf>=5.29.5",  # Override nemo-toolkits constraint of ~=5.29.5
     "setuptools>=80.10.1", # Override setuptools range in other dependencies to address CVE GHSA-58pv-8j8x-9vj2
     "torch==2.11.0; sys_platform == 'linux' and (platform_machine == 'x86_64' or platform_machine == 'aarch64')", # Match vLLM's CUDA requirements; Linux resolves to cu129 via tool.uv.sources
     "torchaudio==2.11.0; sys_platform == 'linux' and (platform_machine == 'x86_64' or platform_machine == 'aarch64')", # Match torch==2.11.0

requirements/text_cuda12-overrides.txt

@@ -3,7 +3,7 @@
 huggingface-hub>=0.34,<1.0
 numba==0.65.0
 numpy>=2.0.0,<=2.2.0
-protobuf>=5.29.5,<7.0
+protobuf>=5.29.5
 setuptools>=80.10.1,<81.0.0
 torch==2.11.0
 torchaudio==2.11.0

tests/backends/ray_data/test_diagnostics.py

@@ -89,9 +89,20 @@ def test_scheduler_diagnostics_are_written_to_ray_session_log(
     ray_data_log = session_dir / "logs" / "ray-data" / "ray-data.log"
 
     assert ray_data_log.exists()
-    log_contents = ray_data_log.read_text()
-    assert {
-        "ray_data_resource_budget_admission",
-        "ray_data_downstream_capacity_admission",
-        "ray_data_actor_autoscaling_decision",
-    } <= set(log_contents.split())
+    event_lines = {
+        event: next(line for line in ray_data_log.read_text().splitlines() if event in line)
+        for event in (
+            "ray_data_resource_budget_admission",
+            "ray_data_downstream_capacity_admission",
+            "ray_data_actor_autoscaling_decision",
+        )
+    }
+
+    for event in ("ray_data_resource_budget_admission", "ray_data_downstream_capacity_admission"):
+        assert "blocked_duration_ms=" in event_lines[event]
+        assert "object_store_internal_bytes=" in event_lines[event]
+        assert "object_store_output_bytes=" in event_lines[event]
+
+    actor_event = event_lines["ray_data_actor_autoscaling_decision"]
+    assert "object_store_internal_bytes=" in actor_event
+    assert "object_store_output_bytes=" in actor_event