Drop columnar frames this process cannot decode

The producers numpy is not the controllers: replicas carry per-deployment
runtime_envs, so a handle can have numpy where the controller does not. Wire-detect
reads the magic bytes and needs no numpy, but decoding does, so such a frame reached
np.frombuffer with np set to None and failed inside ingest -- losing the report with
no explanation and undercounting load at exactly the fan-in sizes this path targets.

Check decodability before decoding and drop the frame with a one-shot warning naming
the cause and the fix.

Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: John Taylor <john.taylor@anyscale.com>

Author: johntaylor-cell

python/ray/serve/_private/autoscaling_metrics_codec.py

@@ -48,8 +48,9 @@
 logger = logging.getLogger(SERVE_LOGGER_NAME)
 
 _MAGIC = b"SCR1"
-# Warn at most once per process that numpy is missing (see should_encode_columnar).
+# Warn at most once per process each (see should_encode_columnar / can_decode_columnar).
 _WARNED_NUMPY_MISSING = False
+_WARNED_NUMPY_UNDECODABLE = False
 
 
 def is_columnar(buf: bytes) -> bool:
@@ -153,6 +154,27 @@ def should_encode_columnar(
     return _widest_metric(report) >= RAY_SERVE_COLUMNAR_METRICS_MIN_REPLICAS
 
 
+def can_decode_columnar() -> bool:
+    """Whether this process can decode columnar frames. The producer's numpy is not
+    the controller's: replicas carry per-deployment runtime_envs, so a frame can
+    arrive here from a process that had numpy when this one does not."""
+    return np is not None
+
+
+def warn_columnar_undecodable_once() -> None:
+    """Warn once per process; a dropped report would otherwise just undercount load."""
+    global _WARNED_NUMPY_UNDECODABLE
+    if _WARNED_NUMPY_UNDECODABLE:
+        return
+    _WARNED_NUMPY_UNDECODABLE = True
+    logger.warning(
+        "Dropping columnar autoscaling metrics: they were sent by a process that has "
+        "numpy, but numpy is not installed here, so they cannot be decoded. Autoscaling "
+        "will undercount load for the affected deployments until numpy is installed in "
+        "this process's environment."
+    )
+
+
 def _widest_metric(report: HandleMetricReport) -> int:
     """Replica count of the report's widest metric, which is what the width gate reads."""
     return max(

python/ray/serve/_private/controller.py

@@ -417,6 +417,10 @@ def record_autoscaling_metrics_from_replica(
         _ingest_start = time.time()
         if isinstance(replica_metric_report, bytes):
             if autoscaling_metrics_codec.is_columnar(replica_metric_report):
+                if not autoscaling_metrics_codec.can_decode_columnar():
+                    # Wire-detect works without numpy but decoding does not.
+                    autoscaling_metrics_codec.warn_columnar_undecodable_once()
+                    return
                 if (
                     RAY_SERVE_AGGREGATE_METRICS_AT_CONTROLLER
                     or RAY_SERVE_ENABLE_DIRECT_INGRESS
@@ -481,6 +485,10 @@ def record_autoscaling_metrics_from_handle(
         _ingest_start = time.time()
         if isinstance(handle_metric_report, bytes):
             if autoscaling_metrics_codec.is_columnar(handle_metric_report):
+                if not autoscaling_metrics_codec.can_decode_columnar():
+                    # Wire-detect works without numpy but decoding does not.
+                    autoscaling_metrics_codec.warn_columnar_undecodable_once()
+                    return
                 if (
                     RAY_SERVE_AGGREGATE_METRICS_AT_CONTROLLER
                     or RAY_SERVE_ENABLE_DIRECT_INGRESS

python/ray/serve/tests/unit/test_columnar_controller_dispatch.py

@@ -146,6 +146,20 @@ def test_handle_cloudpickle_uses_object_store(monkeypatch):
     asm.record_request_metrics_for_handle.assert_called_once()
 
 
+def test_columnar_frame_dropped_when_this_process_lacks_numpy(monkeypatch):
+    """A producer with numpy (replicas get their own runtime_env) can send columnar
+    frames to a controller without it. Wire-detect needs no numpy but decoding does,
+    so the frame must be dropped with a warning rather than raising inside ingest."""
+    monkeypatch.setattr(codec, "np", None)
+    assert codec.can_decode_columnar() is False
+    monkeypatch.setattr(codec, "_WARNED_NUMPY_UNDECODABLE", False)
+    warn = MagicMock()
+    monkeypatch.setattr(codec.logger, "warning", warn)
+    codec.warn_columnar_undecodable_once()
+    codec.warn_columnar_undecodable_once()
+    assert warn.call_count == 1  # once per process, not once per report
+
+
 if __name__ == "__main__":
     import sys