diff --git a/components/src/dynamo/common/multimodal/audio_loader.py b/components/src/dynamo/common/multimodal/audio_loader.py index 719061550768..b9a5941389bd 100644 --- a/components/src/dynamo/common/multimodal/audio_loader.py +++ b/components/src/dynamo/common/multimodal/audio_loader.py @@ -14,6 +14,10 @@ UrlValidationPolicy, validate_media_url, ) +from dynamo.common.multimodal.codec_errors import ( + MissingMediaDecoderError, + audio_decoder_missing, +) from dynamo.common.utils import nvtx_utils as _nvtx from dynamo.common.utils.runtime import run_async @@ -155,6 +159,14 @@ async def load_audio(self, audio_url: str) -> tuple[np.ndarray, float]: # its type and prevent the frontend from returning a 4xx. logger.error("URL rejected loading audio: '%s'", audio_url) raise + except ImportError as exc: + # The image ships no audio decoder (PyAV is deliberately omitted). + # vLLM's own hint here is "pip install vllm[audio]", which drags in + # an unpinned stack; point at the validated bounded install + # instead. NVDEC never decodes audio, so there is no hardware + # alternative to mention. + logger.error("No audio decoder available loading '%s'", audio_url) + raise audio_decoder_missing("vllm", cause=str(exc)) from exc except Exception as exc: logger.error("Error loading audio from %s: %s", audio_url, exc) raise ValueError(f"Failed to load audio from {audio_url}: {exc}") from exc @@ -218,6 +230,7 @@ async def load_audio_batch( collective_exceptions: list[str] = [] status_error: HttpStatusError | None = None url_error: UrlValidationError | None = None + decoder_error: MissingMediaDecoderError | None = None for media_item, result in zip(audio_mm_items, results, strict=True): if isinstance(result, BaseException): if isinstance(result, asyncio.CancelledError): @@ -231,6 +244,10 @@ async def load_audio_batch( status_error = result elif url_error is None and isinstance(result, UrlValidationError): url_error = result + elif decoder_error is None and isinstance( + result, MissingMediaDecoderError + ): + decoder_error = result continue loaded_audio.append(result) @@ -238,6 +255,11 @@ async def load_audio_batch( raise status_error if url_error is not None: raise url_error + if decoder_error is not None: + # Keep the actionable type: the generic aggregate below would erase + # it, and a missing decoder is deployment configuration handlers + # must be able to distinguish from a bad request. + raise decoder_error if collective_exceptions: raise Exception("".join(collective_exceptions)) diff --git a/components/src/dynamo/common/multimodal/codec_errors.py b/components/src/dynamo/common/multimodal/codec_errors.py new file mode 100644 index 000000000000..a5473288a93f --- /dev/null +++ b/components/src/dynamo/common/multimodal/codec_errors.py @@ -0,0 +1,112 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Actionable errors for media the shipped images cannot decode. + +The runtime images deliberately omit the software media decoders (OpenCV, +PyAV, decord) and route H.264/H.265 video to NVDEC hardware decode instead. +Any other input codec needs one of those Python packages, so without an +explicit install the failure surfaces as a bare ``ModuleNotFoundError`` from +deep inside the backend -- no codec, no remedy, and in one observed case the +whole video payload embedded in the message. The builders here name the codec, +the missing package at its validated version bounds, the installer command, +and the hardware alternative, in one place so the three backends cannot drift. + +The version bounds come from +:mod:`dynamo.common.utils.install_media_decoders` (the explicit installer); +importing its constants here is deliberate single-sourcing -- nothing in this +module installs anything. +""" + +from __future__ import annotations + +from dynamo.common.multimodal.nvdec_decoder import HW_ROUTED_CODECS, nvdec_available +from dynamo.common.utils.install_media_decoders import VALIDATED_SPECS + +INSTALLER_CMD = "python -m dynamo.common.utils.install_media_decoders" + + +class MissingMediaDecoderError(RuntimeError): + """A media request needs a decoder package the image does not ship. + + Deliberately not a ``ValueError``: the input may be perfectly valid media. + The gap is deployment configuration, so handlers that map ``ValueError`` + to a client 4xx should not blame the request for it. + """ + + +def _install_hint(backend: str, package: str) -> str: + spec = VALIDATED_SPECS[package] + return ( + f"install the validated decoder with `pip install --no-deps '{spec}'` " + f"(or `{INSTALLER_CMD} {backend}`)" + ) + + +def _with_cause(message: str, cause: str | None) -> str: + """Append the underlying decoder text so diagnostics survive the wrap. + + ``raise ... from exc`` preserves the cause for tracebacks, but handlers + that ship only ``str(exc)`` to the client (HTTP error bodies) would drop + it -- and the underlying reason is part of this error's contract. + """ + if cause: + return f"{message} (decoder reported: {cause})" + return message + + +def video_decoder_missing( + backend: str, + package: str, + module: str, + codec: str | None, + cause: str | None = None, +) -> MissingMediaDecoderError: + """Build the error for a video whose decode path has no decoder. + + Two distinct situations produce it, and the remedy differs: + + * ``codec`` is H.264/H.265 but NVDEC is unavailable in this container -- + the primary fix is granting the ``video`` driver capability, not + installing software. + * any other codec -- NVDEC never decodes it, so the fix is the software + decoder install or re-encoding the input to H.264/H.265. + """ + codec_desc = f"codec '{codec}'" if codec else "an undetected codec" + if codec in HW_ROUTED_CODECS and not nvdec_available(): + lead = ( + f"this video ({codec_desc}) normally decodes in hardware via NVDEC, " + "but NVDEC is unavailable in this container. Grant the 'video' " + "driver capability (NVIDIA_DRIVER_CAPABILITIES) to enable it, or " + ) + else: + lead = ( + f"this video ({codec_desc}) has no decoder in this image: shipped " + "images decode only H.264/H.265 (in hardware, via NVDEC), and the " + f"software decoder '{module}' is deliberately not installed. " + "Re-encode the input to H.264/H.265, or " + ) + return MissingMediaDecoderError( + _with_cause( + "Cannot decode video: " + lead + _install_hint(backend, package) + ".", + cause, + ) + ) + + +def audio_decoder_missing( + backend: str, cause: str | None = None +) -> MissingMediaDecoderError: + """Build the error for audio input with no decoder in the image. + + NVDEC never decodes audio, so unlike video there is no hardware + alternative -- the only remedy is the PyAV install. + """ + return MissingMediaDecoderError( + _with_cause( + "Cannot decode audio: this input needs the PyAV decoder ('av'), which " + "this image deliberately does not ship, and NVDEC does not decode " + "audio. To enable audio input, " + _install_hint(backend, "av") + ".", + cause, + ) + ) diff --git a/components/src/dynamo/common/multimodal/media_source.py b/components/src/dynamo/common/multimodal/media_source.py index cf3f2fdfaf5d..d74ea63bc1ab 100644 --- a/components/src/dynamo/common/multimodal/media_source.py +++ b/components/src/dynamo/common/multimodal/media_source.py @@ -23,6 +23,7 @@ import base64 import binascii import logging +from typing import Final from urllib.parse import unquote, urlparse from urllib.request import url2pathname @@ -44,6 +45,31 @@ def is_local_media_url(url: str) -> bool: return urlparse(url).scheme in LOCAL_MEDIA_SCHEMES +# Longest a media source may render as inside an error message or log line. +# Generous enough to keep an ordinary URL intact and identifiable. +SOURCE_LABEL_LIMIT: Final = 120 + + +def describe_media_source(source: str, limit: int = SOURCE_LABEL_LIMIT) -> str: + """Render ``source`` as a bounded label safe to put in an error or log. + + A ``data:`` URI carries the whole media payload inline, so echoing one into + an error message serializes megabytes of base64 -- to the client, and to + every log sink that records the failure. Describe those by media type and + size instead, never by content. Other sources are truncated, since a URL + identifies the request without being unbounded. + """ + if not isinstance(source, str): + return "" + if source.startswith("data:"): + meta = source[len("data:") :].partition(",")[0] + media_type = meta.split(";")[0] or "application/octet-stream" + return f"data:{media_type} ({len(source)} chars, payload elided)" + if len(source) > limit: + return f"{source[:limit]}... ({len(source)} chars)" + return source + + def _decode_data_uri(url: str) -> bytes: """Decode a ``data:`` URI body to bytes. diff --git a/components/src/dynamo/common/multimodal/nvdec_decoder.py b/components/src/dynamo/common/multimodal/nvdec_decoder.py index 7e411d6685a0..800e6341c2fc 100644 --- a/components/src/dynamo/common/multimodal/nvdec_decoder.py +++ b/components/src/dynamo/common/multimodal/nvdec_decoder.py @@ -24,7 +24,8 @@ Gating: NVDEC is used when PyNvVideoCodec is importable and ``DYN_DISABLE_NVDEC`` is not set. When it is unavailable (CPU image, unsupported profile) the caller -surfaces the actionable "unsupported codec" error. +raises the actionable unsupported-codec error built by +``common.multimodal.codec_errors``. """ from __future__ import annotations diff --git a/components/src/dynamo/common/multimodal/video_loader.py b/components/src/dynamo/common/multimodal/video_loader.py index ca81c29fe981..182baa87a8e9 100644 --- a/components/src/dynamo/common/multimodal/video_loader.py +++ b/components/src/dynamo/common/multimodal/video_loader.py @@ -27,6 +27,10 @@ UrlValidationPolicy, validate_media_url, ) +from dynamo.common.multimodal.codec_errors import ( + MissingMediaDecoderError, + video_decoder_missing, +) from dynamo.common.multimodal.media_source import ( is_local_media_url, read_local_media_bytes, @@ -134,15 +138,7 @@ async def _load_video_with_vllm( content = await fetch_bytes( normalized_url, self._http_timeout, policy=self._url_policy ) - # Dual decode path: hardware-decode H.264/H.265 on NVDEC; every other - # codec falls through to vLLM's software decoder below. Note the - # runtime images purge the software decode wheels (opencv/av/decord/ - # torchcodec) for codec compliance, so that fallback only resolves - # where a decoder has been installed separately. - decoded = await self._maybe_decode_with_nvdec(content) - if decoded is not None: - return decoded - return await asyncio.to_thread(media_io.load_bytes, content) + return await self._decode_video_bytes(content, media_io) # file:// and data: never touch the network, but they still deserve # hardware decode: without this they reach only the software decoder, @@ -152,40 +148,44 @@ async def _load_video_with_vllm( # connector below uses, so this adds no local-read surface. if is_local_media_url(normalized_url): content = await read_local_media_bytes(normalized_url, self._url_policy) - decoded = await self._maybe_decode_with_nvdec(content) - if decoded is not None: - return decoded - return await asyncio.to_thread(media_io.load_bytes, content) + return await self._decode_video_bytes(content, media_io) connector = self._get_vllm_media_connector() return await connector.load_from_url_async( normalized_url, media_io, fetch_timeout=self._http_timeout ) - async def _maybe_decode_with_nvdec( - self, content: bytes - ) -> tuple[np.ndarray, Dict[str, Any]] | None: - """Hardware-decode H.264/H.265 via NVDEC, or None to use the software path. - - Returns None for VP8/VP9/AV1 (handled by the existing decoder), when NVDEC - is unavailable, or when NVDEC fails -- the caller then falls back to the - software decoder, which surfaces the actionable "unsupported codec" error - if it also cannot decode. + async def _decode_video_bytes( + self, content: bytes, media_io: Any + ) -> tuple[np.ndarray, Dict[str, Any]]: + """Decode video bytes: H.264/H.265 on NVDEC, all else software. + + The runtime images purge the software decode wheels (opencv/av/decord/ + torchcodec) for codec compliance, so the software fallback only + resolves where a decoder was installed separately. When it is absent, + vLLM's lazy import surfaces a bare ``No module named 'cv2'`` with no + codec and no remedy -- convert that into the actionable + unsupported-codec error, which can name the codec because the probe + already ran here. """ codec = probe_video_codec(content) - if not should_use_nvdec(codec): - return None + if should_use_nvdec(codec): + try: + return await asyncio.to_thread( + decode_video_nvdec, content, self._num_frames + ) + except Exception as exc: # noqa: BLE001 - fall back to software decode + logger.warning( + "NVDEC decode failed for a %s clip (%s); using software decode", + codec, + exc, + ) try: - return await asyncio.to_thread( - decode_video_nvdec, content, self._num_frames - ) - except Exception as exc: # noqa: BLE001 - fall back to software decode - logger.warning( - "NVDEC decode failed for a %s clip (%s); using software decode", - codec, - exc, - ) - return None + return await asyncio.to_thread(media_io.load_bytes, content) + except ImportError as exc: + raise video_decoder_missing( + "vllm", "opencv-python-headless", "cv2", codec, cause=str(exc) + ) from exc async def load_video(self, video_url: str) -> tuple[np.ndarray, Dict[str, Any]]: try: @@ -203,6 +203,12 @@ async def load_video(self, video_url: str) -> tuple[np.ndarray, Dict[str, Any]]: # its type and prevent the frontend from returning a 4xx. logger.error("URL rejected loading video: '%s'", video_url) raise + except MissingMediaDecoderError: + # Already actionable (names the codec and the install); a missing + # decoder is deployment configuration, not a bad request, so keep + # the type instead of degrading it to the ValueError below. + logger.error("No decoder available for video: '%s'", video_url) + raise except Exception as exc: logger.error("Error loading video from %s: %s", video_url, exc) raise ValueError(f"Failed to load video from {video_url}: {exc}") from exc @@ -249,6 +255,7 @@ async def load_video_batch( collective_exceptions: list[str] = [] status_error: HttpStatusError | None = None url_error: UrlValidationError | None = None + decoder_error: MissingMediaDecoderError | None = None for media_item, result in zip(video_mm_items, results): if isinstance(result, BaseException): if isinstance(result, asyncio.CancelledError): @@ -262,6 +269,10 @@ async def load_video_batch( status_error = result elif url_error is None and isinstance(result, UrlValidationError): url_error = result + elif decoder_error is None and isinstance( + result, MissingMediaDecoderError + ): + decoder_error = result continue frames, metadata = result loaded_videos.append((np.ascontiguousarray(frames), metadata)) @@ -271,6 +282,12 @@ async def load_video_batch( if url_error is not None: raise url_error + if decoder_error is not None: + # Keep the actionable type: the generic aggregate below would erase + # it, and a missing decoder is deployment configuration handlers + # must be able to distinguish from a bad request. + raise decoder_error + if collective_exceptions: raise Exception("".join(collective_exceptions)) diff --git a/components/src/dynamo/common/tests/multimodal/test_audio_loader.py b/components/src/dynamo/common/tests/multimodal/test_audio_loader.py index b3064de1fa15..2052aa6275c4 100644 --- a/components/src/dynamo/common/tests/multimodal/test_audio_loader.py +++ b/components/src/dynamo/common/tests/multimodal/test_audio_loader.py @@ -10,6 +10,8 @@ from dynamo.common.http import HttpStatusError from dynamo.common.http.url_validator import UrlValidationError, UrlValidationPolicy from dynamo.common.multimodal.audio_loader import AudioLoader +from dynamo.common.multimodal.codec_errors import MissingMediaDecoderError +from dynamo.common.utils.install_media_decoders import VALIDATED_SPECS pytestmark = [ pytest.mark.unit, @@ -176,3 +178,36 @@ async def test_load_audio_batch_reads_decoded_variant(monkeypatch): decoded_item, return_metadata=True, ) + + +@pytest.mark.asyncio +async def test_load_audio_missing_decoder_is_actionable(): + """vLLM's own hint here is `pip install vllm[audio]`, which drags in an + unpinned stack; the wrap must point at the validated bounded install and + say there is no hardware alternative for audio.""" + loader = AudioLoader() + loader._load_audio_with_vllm = AsyncMock( # type: ignore[method-assign] + side_effect=ImportError("Please install vllm[audio] for audio support") + ) + + with pytest.raises(MissingMediaDecoderError) as exc_info: + await loader.load_audio("https://example.com/x.mp3") + + msg = str(exc_info.value) + assert VALIDATED_SPECS["av"] in msg + assert "install_media_decoders vllm" in msg + assert "NVDEC does not decode audio" in msg + + +@pytest.mark.asyncio +async def test_load_audio_batch_preserves_missing_decoder_error(): + """The batch aggregate wraps failures in a generic Exception; the + missing-decoder type must survive it (review finding).""" + loader = AudioLoader() + err = audio_loader_module.audio_decoder_missing("vllm") + loader.load_audio = AsyncMock(side_effect=err) # type: ignore[method-assign] + + with pytest.raises(MissingMediaDecoderError) as exc_info: + await loader.load_audio_batch([{"Url": "https://example.com/x.mp3"}]) + + assert exc_info.value is err diff --git a/components/src/dynamo/common/tests/multimodal/test_codec_errors.py b/components/src/dynamo/common/tests/multimodal/test_codec_errors.py new file mode 100644 index 000000000000..5e2d45755a35 --- /dev/null +++ b/components/src/dynamo/common/tests/multimodal/test_codec_errors.py @@ -0,0 +1,106 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for the actionable unsupported-codec error builders.""" + +import pytest + +import dynamo.common.multimodal.codec_errors as codec_errors +from dynamo.common.multimodal.codec_errors import ( + MissingMediaDecoderError, + audio_decoder_missing, + video_decoder_missing, +) +from dynamo.common.utils.install_media_decoders import VALIDATED_SPECS + +pytestmark = [pytest.mark.unit, pytest.mark.pre_merge, pytest.mark.gpu_0] + + +def test_video_message_names_codec_spec_and_installer(monkeypatch): + monkeypatch.setattr(codec_errors, "nvdec_available", lambda: True) + err = video_decoder_missing("vllm", "opencv-python-headless", "cv2", "vp9") + + assert isinstance(err, MissingMediaDecoderError) + msg = str(err) + assert "'vp9'" in msg + assert "cv2" in msg + # The bounded spec comes verbatim from the installer's constants, so the + # message and the documented install can never drift apart. + assert VALIDATED_SPECS["opencv-python-headless"] in msg + assert "install_media_decoders vllm" in msg + # Non-hardware codec: the hardware alternative is re-encoding. + assert "H.264/H.265" in msg + + +def test_hw_codec_without_nvdec_points_at_driver_capability(monkeypatch): + """H.264 with NVDEC unavailable is a capability problem first. + + The primary remedy is granting the container the 'video' driver + capability, not installing a software decoder -- the message must lead + with that. + """ + monkeypatch.setattr(codec_errors, "nvdec_available", lambda: False) + err = video_decoder_missing("vllm", "opencv-python-headless", "cv2", "h264") + + msg = str(err) + assert "NVDEC is unavailable" in msg + assert "NVIDIA_DRIVER_CAPABILITIES" in msg + assert VALIDATED_SPECS["opencv-python-headless"] in msg + + +def test_hw_codec_with_nvdec_available_uses_generic_wording(monkeypatch): + """h264 + NVDEC available but decode still fell through to software: + the capability lead would be wrong, so the generic wording applies.""" + monkeypatch.setattr(codec_errors, "nvdec_available", lambda: True) + err = video_decoder_missing("vllm", "opencv-python-headless", "cv2", "h264") + + assert "NVDEC is unavailable" not in str(err) + + +def test_unknown_codec_still_actionable(monkeypatch): + monkeypatch.setattr(codec_errors, "nvdec_available", lambda: True) + err = video_decoder_missing("sglang", "decord2", "decord", None) + + msg = str(err) + assert "an undetected codec" in msg + assert VALIDATED_SPECS["decord2"] in msg + assert "install_media_decoders sglang" in msg + + +def test_audio_message_has_no_hardware_alternative(): + err = audio_decoder_missing("vllm") + + msg = str(err) + assert isinstance(err, MissingMediaDecoderError) + assert VALIDATED_SPECS["av"] in msg + assert "NVDEC does not decode audio" in msg + assert "install_media_decoders vllm" in msg + + +def test_cause_text_is_appended_and_optional(monkeypatch): + """The underlying decoder text must survive wraps whose handlers ship + only str(exc) to the client; absent a cause, no dangling suffix.""" + monkeypatch.setattr(codec_errors, "nvdec_available", lambda: True) + with_cause = video_decoder_missing( + "vllm", "opencv-python-headless", "cv2", "vp9", cause="No module named 'cv2'" + ) + assert "(decoder reported: No module named 'cv2')" in str(with_cause) + without = video_decoder_missing("vllm", "opencv-python-headless", "cv2", "vp9") + assert "decoder reported" not in str(without) + audio = audio_decoder_missing("vllm", cause="Please install vllm[audio]") + assert "(decoder reported: Please install vllm[audio])" in str(audio) + + +def test_error_is_not_a_value_error(): + """Handlers map ValueError to client 4xx; a missing decoder is deployment + configuration and must not be blamed on the request.""" + assert not issubclass(MissingMediaDecoderError, ValueError) + assert issubclass(MissingMediaDecoderError, RuntimeError) + + +def test_every_referenced_package_has_a_validated_spec(): + """The builders promise a bounded spec for these packages; keep the + installer's table covering them.""" + for package in ("opencv-python-headless", "decord2", "av"): + spec = VALIDATED_SPECS[package] + assert ">=" in spec and ",<" in spec diff --git a/components/src/dynamo/common/tests/multimodal/test_media_source.py b/components/src/dynamo/common/tests/multimodal/test_media_source.py index 63180e7400f1..6aa74eea37b8 100644 --- a/components/src/dynamo/common/tests/multimodal/test_media_source.py +++ b/components/src/dynamo/common/tests/multimodal/test_media_source.py @@ -18,6 +18,7 @@ from dynamo.common.http.url_validator import UrlValidationError, UrlValidationPolicy from dynamo.common.multimodal.media_source import ( + describe_media_source, is_local_media_url, read_local_media_bytes, ) @@ -110,3 +111,35 @@ async def test_malformed_data_uri_rejected(url, match): async def test_unsupported_scheme_rejected(): with pytest.raises(UrlValidationError, match="Unsupported local media scheme"): await read_local_media_bytes("s3://bucket/clip.mp4", UrlValidationPolicy()) + + +def test_describe_media_source_elides_a_data_uri_payload() -> None: + """A data: URI is the whole media payload; describing one must never + reproduce it, only its type and size.""" + payload = "A" * 100_000 + label = describe_media_source(f"data:video/mp4;base64,{payload}") + + assert payload not in label + assert "data:video/mp4" in label + assert "payload elided" in label + assert len(label) < 100 + + +def test_describe_media_source_keeps_an_ordinary_url_intact() -> None: + url = "https://example.com/clip.mp4" + assert describe_media_source(url) == url + + +def test_describe_media_source_bounds_an_overlong_url() -> None: + url = "https://example.com/" + "a" * 500 + label = describe_media_source(url) + + assert label.startswith("https://example.com/") + assert len(label) < len(url) + assert str(len(url)) in label # keeps the true size visible + + +def test_describe_media_source_survives_a_non_string() -> None: + """The video loop labels whatever it was handed, including malformed + items, before it has established the input is a string.""" + assert describe_media_source(None) == "" # type: ignore[arg-type] diff --git a/components/src/dynamo/common/tests/multimodal/test_video_loader.py b/components/src/dynamo/common/tests/multimodal/test_video_loader.py index db26ac272d38..d07f5659693b 100644 --- a/components/src/dynamo/common/tests/multimodal/test_video_loader.py +++ b/components/src/dynamo/common/tests/multimodal/test_video_loader.py @@ -9,7 +9,9 @@ import dynamo.common.multimodal.video_loader as video_loader_module from dynamo.common.http import HttpStatusError from dynamo.common.http.url_validator import UrlValidationError, UrlValidationPolicy +from dynamo.common.multimodal.codec_errors import MissingMediaDecoderError from dynamo.common.multimodal.video_loader import VideoLoader +from dynamo.common.utils.install_media_decoders import VALIDATED_SPECS pytestmark = [ pytest.mark.unit, @@ -154,7 +156,7 @@ async def test_load_video_batch_reads_decoded_variant_with_metadata(monkeypatch) @pytest.mark.asyncio -async def test_maybe_decode_with_nvdec_routes_h264(monkeypatch): +async def test_decode_video_bytes_routes_h264_to_nvdec(monkeypatch): loader = VideoLoader() frames = np.zeros((2, 4, 6, 3), dtype=np.uint8) meta = {"fps": 30.0, "frames_indices": [0, 1], "total_num_frames": 2} @@ -163,18 +165,19 @@ async def test_maybe_decode_with_nvdec_routes_h264(monkeypatch): monkeypatch.setattr( video_loader_module, "decode_video_nvdec", lambda b, n: (frames, meta) ) + media_io = _RecordingMediaIO(frames) - result = await loader._maybe_decode_with_nvdec(b"h264-bytes") + got_frames, got_meta = await loader._decode_video_bytes(b"h264-bytes", media_io) - assert result is not None - got_frames, got_meta = result np.testing.assert_array_equal(got_frames, frames) assert got_meta == meta + assert media_io.calls == [] # hardware decoded; software untouched @pytest.mark.asyncio -async def test_maybe_decode_with_nvdec_skips_royalty_free(monkeypatch): +async def test_decode_video_bytes_routes_royalty_free_to_software(monkeypatch): loader = VideoLoader() + frames = np.zeros((1, 2, 2, 3), dtype=np.uint8) monkeypatch.setattr(video_loader_module, "probe_video_codec", lambda b: "vp9") monkeypatch.setattr(video_loader_module, "should_use_nvdec", lambda c: False) called = {"decode": False} @@ -183,16 +186,19 @@ def _decode(*a, **k): called["decode"] = True monkeypatch.setattr(video_loader_module, "decode_video_nvdec", _decode) + media_io = _RecordingMediaIO(frames) - result = await loader._maybe_decode_with_nvdec(b"vp9-bytes") + result = await loader._decode_video_bytes(b"vp9-bytes", media_io) - assert result is None # VP9 stays on the software path + assert result is media_io.result # VP9 stays on the software path assert called["decode"] is False # NVDEC not invoked + assert media_io.calls == [b"vp9-bytes"] @pytest.mark.asyncio -async def test_maybe_decode_with_nvdec_falls_back_on_failure(monkeypatch): +async def test_decode_video_bytes_falls_back_on_nvdec_failure(monkeypatch): loader = VideoLoader() + frames = np.zeros((1, 2, 2, 3), dtype=np.uint8) monkeypatch.setattr(video_loader_module, "probe_video_codec", lambda b: "hevc") monkeypatch.setattr(video_loader_module, "should_use_nvdec", lambda c: True) @@ -200,7 +206,87 @@ def _boom(*a, **k): raise RuntimeError("nvdec session limit") monkeypatch.setattr(video_loader_module, "decode_video_nvdec", _boom) + media_io = _RecordingMediaIO(frames) - result = await loader._maybe_decode_with_nvdec(b"hevc-bytes") + result = await loader._decode_video_bytes(b"hevc-bytes", media_io) - assert result is None # a NVDEC failure falls back, never raises + assert result is media_io.result # NVDEC failure falls back, never raises + assert media_io.calls == [b"hevc-bytes"] + + +class _RecordingMediaIO: + """Stub for vLLM's VideoMediaIO: records load_bytes calls.""" + + def __init__(self, frames): + self.result = (frames, {"fps": 1.0}) + self.calls: list[bytes] = [] + + def load_bytes(self, content: bytes): + self.calls.append(content) + return self.result + + +class _ImportErrorMediaIO: + """Stub reproducing vLLM's lazy cv2 import failure on a stripped image.""" + + def load_bytes(self, content: bytes): + raise ModuleNotFoundError("No module named 'cv2'", name="cv2") + + +@pytest.mark.asyncio +async def test_decode_video_bytes_missing_decoder_is_actionable(monkeypatch): + """A bare `No module named 'cv2'` must become the actionable codec error. + + Reproduced on a real runtime image before this existed: the user-visible + text was `Failed to load video from ...: No module named 'cv2'` -- no + codec, no remedy. + """ + loader = VideoLoader() + monkeypatch.setattr(video_loader_module, "probe_video_codec", lambda b: "vp9") + monkeypatch.setattr(video_loader_module, "should_use_nvdec", lambda c: False) + + with pytest.raises(MissingMediaDecoderError) as exc_info: + await loader._decode_video_bytes(b"vp9-bytes", _ImportErrorMediaIO()) + + msg = str(exc_info.value) + assert "'vp9'" in msg # names the codec + assert VALIDATED_SPECS["opencv-python-headless"] in msg # bounded spec + assert "install_media_decoders vllm" in msg # installer command + assert "cv2" in msg + + +@pytest.mark.asyncio +async def test_load_video_batch_preserves_missing_decoder_error(): + """The batch aggregate wraps failures in a generic Exception; the + missing-decoder type must survive it (review finding).""" + loader = VideoLoader() + err = video_loader_module.video_decoder_missing( + "vllm", "opencv-python-headless", "cv2", "vp9" + ) + loader.load_video = AsyncMock(side_effect=err) # type: ignore[method-assign] + + with pytest.raises(MissingMediaDecoderError) as exc_info: + await loader.load_video_batch([{"Url": "https://example.com/x.mp4"}]) + + assert exc_info.value is err + + +@pytest.mark.asyncio +async def test_load_video_preserves_missing_decoder_error(monkeypatch): + """The generic ValueError wrap must not erase the decoder-missing type. + + A missing decoder is deployment configuration, not a bad request, so it + must not degrade into the ValueError that handlers map to a client error. + """ + loader = VideoLoader() + err = video_loader_module.video_decoder_missing( + "vllm", "opencv-python-headless", "cv2", "vp9" + ) + loader._load_video_with_vllm = AsyncMock( # type: ignore[method-assign] + side_effect=err + ) + + with pytest.raises(MissingMediaDecoderError) as exc_info: + await loader.load_video("https://example.com/x.mp4") + + assert exc_info.value is err diff --git a/components/src/dynamo/common/tests/test_install_media_decoders.py b/components/src/dynamo/common/tests/test_install_media_decoders.py index 69008586a6c2..03576888e075 100644 --- a/components/src/dynamo/common/tests/test_install_media_decoders.py +++ b/components/src/dynamo/common/tests/test_install_media_decoders.py @@ -441,41 +441,138 @@ def test_cli_pip_args_reach_pip(sandboxed): # Nothing implicit: the env-var/startup pathway must not creep back. # --------------------------------------------------------------------------- -_COMPONENTS_ROOT = Path(__file__).resolve().parents[3] +# The dynamo package root (parents[2] = .../dynamo). parents[3] would be +# components/src in the repo but site-packages in an installed layout, where +# the sweep then walks every third-party package -- including files with +# Python 2 syntax that ast.parse cannot read. +_PACKAGE_ROOT = Path(__file__).resolve().parents[2] -def test_no_production_code_invokes_the_installer(): +def test_no_entrypoint_references_the_installer(): """The installer is operator-run only. Reviewers rejected implicit installation at worker startup (env-var gated hooks in entrypoints): an install that changes the container's codec - surface has to be a visible, deliberate step. This sweep keeps any - `__main__.py` from calling the installer and the retired env switch from - coming back anywhere under components/src. + surface has to be a visible, deliberate step. Entrypoints are where + startup happens, so no `__main__.py` may reference the installer at all. + + Other production code MAY import its constants -- the actionable + unsupported-codec errors single-source their version bounds from + VALIDATED_SPECS -- so this sweep is scoped to entrypoints, and the two + sweeps below cover the rest: nothing may CALL the installer, and the + retired env switch must not come back anywhere. + """ + offenders: list[str] = [] + for path in sorted(_PACKAGE_ROOT.rglob("__main__.py")): + text = path.read_text(encoding="utf-8") + if "install_media_decoders" in text or "DYN_ENABLE_MEDIA_DECODERS" in text: + offenders.append(str(path.relative_to(_PACKAGE_ROOT))) + assert not offenders, ( + f"{offenders} reference the media-decoder installer from an entrypoint; " + "it must stay an explicit operator command, never wired into startup" + ) + + +def _source_calls_installer(source: str) -> bool: + """AST-based detection of a call into the installer, aliases included. + + A literal `install_media_decoders(` grep misses + `from ... import install_media_decoders as x; x()` and + `import ...install_media_decoders as m; m.main()`. Walk the AST instead: + collect every name the installer module (or its functions) is bound to, + then flag any Call through one of those bindings. + """ + import ast + + try: + tree = ast.parse(source) + except SyntaxError: + # Not parseable as this interpreter's Python (vendored/py2-era file); + # it cannot be importing our installer through the import system. + return False + fn_aliases: set[str] = set() # names bound to installer functions + mod_aliases: set[str] = set() # names bound to the installer module + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.module: + if node.module.endswith("install_media_decoders"): + for a in node.names: + if a.name in ("install_media_decoders", "main"): + fn_aliases.add(a.asname or a.name) + elif node.module.endswith("common.utils") or node.module == "utils": + for a in node.names: + if a.name == "install_media_decoders": + mod_aliases.add(a.asname or a.name) + elif isinstance(node, ast.Import): + for a in node.names: + if a.name.endswith("install_media_decoders"): + mod_aliases.add(a.asname or a.name.split(".")[0]) + if not fn_aliases and not mod_aliases: + return False + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + f = node.func + if isinstance(f, ast.Name) and f.id in fn_aliases: + return True + if isinstance(f, ast.Attribute) and f.attr in ( + "install_media_decoders", + "main", + ): + root = f.value + while isinstance(root, ast.Attribute): + root = root.value + if isinstance(root, ast.Name) and root.id in mod_aliases: + return True + return False + + +def test_no_production_code_calls_the_installer(): + """Importing constants is fine; invoking the install is not. + + Calls into the installer (through any import alias) and the retired + DYN_ENABLE_MEDIA_DECODERS switch must appear nowhere outside the + installer module and its test. """ allowed = { - Path("dynamo/common/utils/install_media_decoders.py"), # the installer itself - Path("dynamo/common/tests/test_install_media_decoders.py"), # this test + Path("common/utils/install_media_decoders.py"), # the installer itself + Path("common/tests/test_install_media_decoders.py"), # this test } offenders: list[str] = [] - for path in sorted(_COMPONENTS_ROOT.rglob("*.py")): - rel = path.relative_to(_COMPONENTS_ROOT) + for path in sorted(_PACKAGE_ROOT.rglob("*.py")): + rel = path.relative_to(_PACKAGE_ROOT) if rel in allowed: continue text = path.read_text(encoding="utf-8") - if "install_media_decoders" in text or "DYN_ENABLE_MEDIA_DECODERS" in text: + if "DYN_ENABLE_MEDIA_DECODERS" in text or _source_calls_installer(text): offenders.append(str(rel)) assert not offenders, ( - f"{offenders} reference the media-decoder installer; it must stay an " - "explicit operator command, never wired into worker startup" + f"{offenders} invoke the media-decoder installer (or resurrect its env " + "switch); only an operator may run it" ) +def test_call_detector_sees_through_aliases(): + """The detector must catch aliased calls and ignore constant imports.""" + calls = _source_calls_installer + direct = "from dynamo.common.utils.install_media_decoders import install_media_decoders\ninstall_media_decoders('vllm')\n" + aliased = "from dynamo.common.utils.install_media_decoders import install_media_decoders as x\nx('vllm')\n" + mod_alias = ( + "import dynamo.common.utils.install_media_decoders as m\nm.main(['vllm'])\n" + ) + from_pkg = "from dynamo.common.utils import install_media_decoders as imd\nimd.install_media_decoders('vllm')\n" + constants_only = "from dynamo.common.utils.install_media_decoders import VALIDATED_SPECS\nprint(VALIDATED_SPECS)\n" + assert calls(direct) + assert calls(aliased) + assert calls(mod_alias) + assert calls(from_pkg) + assert not calls(constants_only) + + def test_installer_module_has_no_env_switches(): """The module reads no environment variables at all.""" - source = ( - _COMPONENTS_ROOT / "dynamo/common/utils/install_media_decoders.py" - ).read_text(encoding="utf-8") + source = (_PACKAGE_ROOT / "common/utils/install_media_decoders.py").read_text( + encoding="utf-8" + ) assert "os.environ" not in source and "getenv" not in source, ( "install_media_decoders.py reads the environment; configuration belongs in " "CLI flags so the install stays explicit and self-describing" diff --git a/components/src/dynamo/sglang/request_handlers/multimodal/encode_worker_handler.py b/components/src/dynamo/sglang/request_handlers/multimodal/encode_worker_handler.py index c197a22d1d84..f790ea0ef1c0 100644 --- a/components/src/dynamo/sglang/request_handlers/multimodal/encode_worker_handler.py +++ b/components/src/dynamo/sglang/request_handlers/multimodal/encode_worker_handler.py @@ -36,6 +36,10 @@ ) from dynamo.common.multimodal import EMBEDDING_SENDER_FACTORIES, ImageLoader from dynamo.common.multimodal.cache_uuid import reject_unsupported_multimodal_uuids +from dynamo.common.multimodal.codec_errors import ( + MissingMediaDecoderError, + video_decoder_missing, +) from dynamo.common.multimodal.image_loader import DECODED_VARIANT_KEY, URL_VARIANT_KEY from dynamo.common.multimodal.media_descriptor import decoded_content_hash_key from dynamo.common.multimodal.media_source import ( @@ -98,6 +102,18 @@ class _ModalityBatch: url_attr: str +def _software_video_decoder_imports() -> bool: + """True when SGLang's software video decoder (torchcodec or decord) + actually imports -- not merely resolves to a spec.""" + for module in ("torchcodec", "decord"): + try: + importlib.import_module(module) + return True + except Exception: # noqa: BLE001 - broken installs count as absent + continue + return False + + # SGLang model types whose video preprocessing needs per-frame timestamps from # ``video_metadata``. For these, ``_process_video_items`` runs # ``for m in video_metadata: m.get("fps")`` (sglang.srt.disaggregation.encode_server, @@ -604,9 +620,26 @@ async def _maybe_nvdec_decoder(self, url: str) -> Optional[Any]: content = await read_local_media_bytes(normalized, self._url_policy) else: return None - if not should_use_nvdec(probe_video_codec(content)): - # Not a hardware codec, but the bytes are already here and were - # fetched under policy. Hand them over instead of the URL. + codec = probe_video_codec(content) + if not should_use_nvdec(codec): + # Not going to hardware. SGLang's software path needs + # torchcodec or decord, which the codec-compliant image strips + # -- without this preflight the failure happens deep inside + # SGLang as a bare "No module named 'decord'" with the whole + # video payload repr embedded in the message. Fail here, where + # the codec is known and the message can be actionable. Runs + # only for an already-validated video URL, after fetch, so + # payload-validation errors keep precedence. + # + # Real import, not find_spec: a package whose files exist but + # whose native libraries cannot load has a spec and would pass + # a find_spec preflight only to fail deep in SGLang anyway. + # Success is cached in sys.modules, so the cost is first + # request only. + if not _software_video_decoder_imports(): + raise video_decoder_missing("sglang", "decord2", "decord", codec) + # A software decoder exists; the bytes are already here and + # were fetched under policy. Hand them over instead of the URL. return content # Constructing the decoder opens the container and reads its frame # index, so keep it off the event loop. @@ -627,11 +660,25 @@ async def _maybe_nvdec_decoder(self, url: str) -> Optional[Any]: # handler already reports a bad request, so the caller surfaces it # as one instead of silently widening what the deployment accepts. raise + except MissingMediaDecoderError: + # The preflight above is the actionable error this path exists to + # raise. Letting the broad handler below catch it would return the + # bytes anyway and reproduce exactly the deep-SGLang failure it + # replaces. + raise except Exception as exc: # noqa: BLE001 - additive; never blocks the path # If the fetch itself failed there are no bytes and the URL is all # the caller has. If it succeeded and only the decoder construction # failed, pass the validated bytes on rather than making SGLang - # fetch them again. + # fetch them again -- but only if SGLang can actually decode them: + # this fallback leg reaches SGLang's software path exactly like the + # non-hardware-codec leg above, so it needs the same preflight, or + # a host with broken NVDEC and no software decoder gets the deep + # payload-blob error back. + if content is not None and not _software_video_decoder_imports(): + raise video_decoder_missing( + "sglang", "decord2", "decord", probe_video_codec(content) + ) from exc logger.warning( "NVDEC decode failed for video URL (%s); falling back to %s", exc, @@ -649,15 +696,41 @@ async def _build_encode_inputs( ``NvdecVideoDecoder`` (H.264/H.265), the fetched bytes (any other codec, so SGLang does not re-download what we already validated and hold), or the original URL string when nothing was fetched. - Non-video modalities, decoded inputs, and disabled/ineligible cases - are returned unchanged. + Non-video modalities and decoded inputs are returned unchanged. When + NVDEC is disabled or ineligible the URLs are returned policy-validated + and normalized, since SGLang fetches them with its own session. Called from both the cached and uncached encode paths. The embedding cache is disabled by default, so routing this only through the cached path would leave hardware decode unreachable in a stock deployment. """ - if modality_name != "VIDEO" or not self._nvdec_video_enabled(): + if modality_name != "VIDEO": return media_inputs + if not self._nvdec_video_enabled(): + # NVDEC off (CPU image, DYN_DISABLE_NVDEC, or a gated model type): + # these URLs go straight to SGLang's software path, which fetches + # them with its own session and never consults our url policy. Run + # the policy here so a source we would refuse is refused before + # SGLang can reach it -- and before we answer with anything about + # this deployment, since a request we reject is not the place to + # report which decoders are installed. + validated = [ + await validate_media_url(media_input, self._url_policy) + if isinstance(media_input, str) + else media_input + for media_input in media_inputs + ] + # Without this preflight these deployments -- the ones MOST likely + # to lack a decoder entirely -- still get the deep + # "No module named 'decord'" with the payload repr embedded. No + # bytes were fetched here, so the codec cannot be named. Only str + # items count: pre-decoded frontend variants need no decoder. + if ( + any(isinstance(media_input, str) for media_input in media_inputs) + and not _software_video_decoder_imports() + ): + raise video_decoder_missing("sglang", "decord2", "decord", None) + return validated encode_inputs: list[Any] = [] for media_input in media_inputs: if not isinstance(media_input, str): diff --git a/components/src/dynamo/sglang/tests/test_sglang_multimodal_embedding_cache.py b/components/src/dynamo/sglang/tests/test_sglang_multimodal_embedding_cache.py index 2a89d37ba45e..69fa11a9a01c 100644 --- a/components/src/dynamo/sglang/tests/test_sglang_multimodal_embedding_cache.py +++ b/components/src/dynamo/sglang/tests/test_sglang_multimodal_embedding_cache.py @@ -11,6 +11,11 @@ import pytest import torch +from dynamo.common.http.url_validator import ( + UrlValidationError, + UrlValidationPolicy, + validate_media_url, +) from dynamo.common.memory.multimodal_embedding_cache_manager import ( CachedEmbedding, MultimodalEmbeddingCacheManager, @@ -26,6 +31,10 @@ pytest.mark.sglang, pytest.mark.multimodal, pytest.mark.gpu_0, + # These are sub-second unit tests. A generous cap so a hang here fails this + # test instead of stalling the whole session: an unstubbed network call once + # cost ~400s of teardown and took the CI container down with it. + pytest.mark.timeout(60), pytest.mark.profiled_vram_gib(0), pytest.mark.pre_merge, pytest.mark.skipif(Modality is None, reason="SGLang Modality is required"), @@ -33,8 +42,42 @@ @pytest.fixture -def cache_handler() -> MultimodalEncodeWorkerHandler: - """Create a lightweight handler instance for cache-path unit tests.""" +def cache_handler(monkeypatch) -> MultimodalEncodeWorkerHandler: + """Create a lightweight handler instance for cache-path unit tests. + + Default test-world assumption: a software video decoder exists, so the + URL/bytes flow under test is reachable -- the codec-compliant test image + ships none, and without this stub the handler's decoder preflight fires + before the cache logic these tests exercise. Tests about decoder ABSENCE + override the stub explicitly. + + URL validation is stubbed out for the same reason: it resolves the + hostname (``loop.getaddrinfo``) to check the address against the blocked + ranges, and the CPU test container has no DNS, so a real lookup blocks + until the job is killed rather than failing. Tests about validation + itself restore the real function. + """ + monkeypatch.setattr(f"{_HANDLER_MOD}._software_video_decoder_imports", lambda: True) + + async def _passthrough_url(url, _policy): + return url + + monkeypatch.setattr(f"{_HANDLER_MOD}.validate_media_url", _passthrough_url) + + # No unit test may reach the network. Giving the handler a real url policy + # (below) removed the AttributeError that used to abort _maybe_nvdec_decoder + # before it fetched, so on an NVDEC-capable host these cache tests started + # issuing a real request for their example.com URL: ~400s of teardown while + # the event loop waited on the connection, which killed the CI job. The + # broad except in _maybe_nvdec_decoder turns this into the URL passthrough + # the cache tests already expected; tests that exercise fetching stub it + # with their own payload. + async def _no_network(*_args, **_kwargs): + raise AssertionError( + "unit tests must not fetch over the network; stub fetch_bytes" + ) + + monkeypatch.setattr(f"{_HANDLER_MOD}.fetch_bytes", _no_network) class _DummyEncoder: def __init__(self) -> None: @@ -64,6 +107,9 @@ def _set_token_ids_for_test(image_token_id: int, video_token_id: int) -> None: capacity_bytes=32 * 1024 * 1024 ) handler._cache_publisher = None + # The real __init__ always sets this (UrlValidationPolicy.from_env); the + # video paths read it, so a handler built with __new__ needs it too. + handler._url_policy = UrlValidationPolicy() handler.encoder = _DummyEncoder() return handler @@ -459,14 +505,59 @@ async def test_video_cache_key_includes_sampling_config( _HANDLER_MOD = "dynamo.sglang.request_handlers.multimodal.encode_worker_handler" +# A data: URI passes the url policy without touching the network, so the +# decoder-gating tests below exercise gating rather than DNS. +_INLINE_VIDEO = "data:video/mp4;base64,AAAAIGZ0eXBpc29t" +# http:// is refused on scheme alone (allow_http defaults False), so the +# validation tests are deterministic and offline too. +_BLOCKED_URL = "http://169.254.169.254/latest/meta-data/" + + @pytest.fixture def nvdec_handler(cache_handler) -> MultimodalEncodeWorkerHandler: """cache_handler wired with the attributes the NVDEC path reads.""" cache_handler.num_video_frames = 32 - cache_handler._url_policy = SimpleNamespace() return cache_handler +@pytest.mark.asyncio +async def test_disabled_nvdec_without_software_decoder_is_actionable( + nvdec_handler, monkeypatch +) -> None: + """NVDEC off (env/CPU/gated model) + no software decoder: the URLs would + go straight to SGLang and die deep with the payload-blob error, so + _build_encode_inputs must raise the actionable error up front. + + This is the deployment class MOST likely to lack a decoder entirely -- + reviewers caught that the preflight originally lived only on the + NVDEC-enabled path and never ran here. + """ + from dynamo.common.multimodal.codec_errors import MissingMediaDecoderError + + monkeypatch.setenv("DYN_DISABLE_NVDEC", "1") + monkeypatch.setattr( + f"{_HANDLER_MOD}._software_video_decoder_imports", lambda: False + ) + + with pytest.raises(MissingMediaDecoderError) as exc_info: + await nvdec_handler._build_encode_inputs([_INLINE_VIDEO], "VIDEO") + + assert "install_media_decoders sglang" in str(exc_info.value) + + +@pytest.mark.asyncio +async def test_disabled_nvdec_with_software_decoder_passes_urls( + nvdec_handler, monkeypatch +) -> None: + """NVDEC off but a software decoder exists: URLs pass through unchanged + (SGLang fetches and decodes them itself, as before).""" + monkeypatch.setenv("DYN_DISABLE_NVDEC", "1") + monkeypatch.setattr(f"{_HANDLER_MOD}._software_video_decoder_imports", lambda: True) + + out = await nvdec_handler._build_encode_inputs([_INLINE_VIDEO], "VIDEO") + assert out == [_INLINE_VIDEO] + + def test_nvdec_video_enabled_gating(nvdec_handler, monkeypatch) -> None: monkeypatch.setattr(f"{_HANDLER_MOD}.nvdec_available", lambda: True) @@ -584,6 +675,11 @@ async def test_maybe_nvdec_decoder_returns_bytes_for_non_hw_codec( monkeypatch.setattr(f"{_HANDLER_MOD}.fetch_bytes", fetch) monkeypatch.setattr(f"{_HANDLER_MOD}.probe_video_codec", lambda _b: "vp9") monkeypatch.setattr(f"{_HANDLER_MOD}.should_use_nvdec", lambda c: c == "h264") + # The passthrough contract now holds only when SGLang can actually decode + # the bytes; the codec-compliant test image ships no software decoder, so + # stub the preflight probe. The absent-decoder leg (actionable error) is + # covered in test_sglang_multimodal_video.py. + monkeypatch.setattr(f"{_HANDLER_MOD}._software_video_decoder_imports", lambda: True) out = await nvdec_handler._maybe_nvdec_decoder("https://x/clip.webm") @@ -699,11 +795,76 @@ def _boom(_data): raise RuntimeError("NVDEC init failed") monkeypatch.setattr(f"{_HANDLER_MOD}.NvdecVideoDecoder", _boom) + # Bytes fallback is only valid when SGLang can decode them; stub the + # preflight probe (the image ships no software decoder). + monkeypatch.setattr(f"{_HANDLER_MOD}._software_video_decoder_imports", lambda: True) assert await nvdec_handler._maybe_nvdec_decoder("https://x/clip.mp4") == b"bytes" - out = await nvdec_handler._build_encode_inputs(["https://x/clip.mp4"], "VIDEO") + out = await nvdec_handler._build_encode_inputs([_INLINE_VIDEO], "VIDEO") assert out == [b"bytes"] +@pytest.mark.asyncio +@pytest.mark.parametrize("software_decoder_present", [False, True]) +async def test_disabled_nvdec_validates_url_before_reporting_decoders( + nvdec_handler, monkeypatch, software_decoder_present +) -> None: + """A source the policy refuses must be refused here, whatever the decoders. + + With NVDEC off, SGLang fetches these URLs with its own session and never + consults our policy, so this is the only place the policy can apply. Two + failures this pins, both reproduced on the real image: a blocked URL was + handed back for SGLang to fetch when a decoder was present, and answered + with "install decord2" -- deployment configuration, in response to a + request we should refuse -- when one was not. + """ + monkeypatch.setenv("DYN_DISABLE_NVDEC", "1") + monkeypatch.setattr( + f"{_HANDLER_MOD}._software_video_decoder_imports", + lambda: software_decoder_present, + ) + # Undo the fixture's passthrough: validation is what this test asserts. + # The URL below is refused on scheme, so this still performs no lookup. + monkeypatch.setattr(f"{_HANDLER_MOD}.validate_media_url", validate_media_url) + + with pytest.raises(UrlValidationError): + await nvdec_handler._build_encode_inputs([_BLOCKED_URL], "VIDEO") + + +@pytest.mark.asyncio +async def test_decode_failure_without_software_decoder_is_actionable( + nvdec_handler, monkeypatch +) -> None: + """NVDEC failed AND no software decoder exists: the bytes would only die + deep inside SGLang with the payload repr in the message, so the fallback + must raise the actionable error instead of passing them on.""" + from dynamo.common.multimodal.codec_errors import MissingMediaDecoderError + + monkeypatch.setattr(f"{_HANDLER_MOD}.nvdec_available", lambda: True) + nvdec_handler.encoder.model_type = "qwen2_5_vl" + monkeypatch.setattr( + f"{_HANDLER_MOD}.validate_media_url", + AsyncMock(return_value="https://x/clip.mp4"), + ) + monkeypatch.setattr(f"{_HANDLER_MOD}.fetch_bytes", AsyncMock(return_value=b"bytes")) + monkeypatch.setattr(f"{_HANDLER_MOD}.probe_video_codec", lambda _b: "h264") + monkeypatch.setattr(f"{_HANDLER_MOD}.should_use_nvdec", lambda _c: True) + + def _boom(_data): + raise RuntimeError("NVDEC init failed") + + monkeypatch.setattr(f"{_HANDLER_MOD}.NvdecVideoDecoder", _boom) + monkeypatch.setattr( + f"{_HANDLER_MOD}._software_video_decoder_imports", lambda: False + ) + + with pytest.raises(MissingMediaDecoderError) as exc_info: + await nvdec_handler._maybe_nvdec_decoder("https://x/clip.mp4") + + # h264 + NVDEC "available" but failing: the message still leads with the + # capability/hardware framing and carries the install remedy. + assert "install_media_decoders sglang" in str(exc_info.value) + + @pytest.mark.asyncio async def test_video_routes_through_nvdec_with_cache_disabled( nvdec_handler, monkeypatch diff --git a/components/src/dynamo/sglang/tests/test_sglang_multimodal_video.py b/components/src/dynamo/sglang/tests/test_sglang_multimodal_video.py index b84f6d8ffe81..a62e2258f1f8 100644 --- a/components/src/dynamo/sglang/tests/test_sglang_multimodal_video.py +++ b/components/src/dynamo/sglang/tests/test_sglang_multimodal_video.py @@ -1,6 +1,8 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import importlib + import numpy as np import pytest import torch @@ -187,3 +189,96 @@ async def test_nvdec_video_metadata_shim_stamps_valid_metadata(): assert meta_none is None finally: es.preprocess_video = saved + + +# --------------------------------------------------------------------------- +# Unsupported-codec preflight in _maybe_nvdec_decoder. +# --------------------------------------------------------------------------- + + +def _bare_handler(): + """Handler skeleton for exercising _maybe_nvdec_decoder in isolation.""" + from dynamo.common.http.url_validator import UrlValidationPolicy + from dynamo.sglang.request_handlers.multimodal.encode_worker_handler import ( + MultimodalEncodeWorkerHandler, + ) + + handler = object.__new__(MultimodalEncodeWorkerHandler) + handler._url_policy = UrlValidationPolicy.from_env() + return handler + + +def _selective_import(present: set[str]): + """Fake importlib.import_module for the preflight's real-import probe: + decoder modules import only when listed in `present` (a broken native + install behaves exactly like an absent one -- ImportError either way).""" + real = importlib.import_module + + def fake(name, *args, **kwargs): + if name in ("torchcodec", "decord"): + if name in present: + return object() + raise ImportError(f"No module named '{name}' (or broken install)") + return real(name, *args, **kwargs) + + return fake + + +@pytest.mark.asyncio +async def test_vp9_without_software_decoder_is_actionable(monkeypatch): + """A VP9 request on an image with neither torchcodec nor decord must fail + HERE with guidance, not deep inside SGLang with the video payload repr + embedded in a bare "No module named 'decord'" (observed on a real image). + + Raising through _maybe_nvdec_decoder also proves the broad fallback + except-clause does not swallow the preflight error and pass the bytes on + anyway. + """ + import dynamo.sglang.request_handlers.multimodal.encode_worker_handler as ewh + from dynamo.common.multimodal.codec_errors import MissingMediaDecoderError + from dynamo.common.utils.install_media_decoders import VALIDATED_SPECS + + async def fake_validate(url, policy): + return url + + async def fake_fetch(url, timeout, policy=None): + return b"vp9-bytes" + + monkeypatch.setattr(ewh, "validate_media_url", fake_validate) + monkeypatch.setattr(ewh, "fetch_bytes", fake_fetch) + monkeypatch.setattr(ewh, "probe_video_codec", lambda b: "vp9") + monkeypatch.setattr(ewh, "should_use_nvdec", lambda c: False) + monkeypatch.setattr(ewh.importlib, "import_module", _selective_import(set())) + + handler = _bare_handler() + with pytest.raises(MissingMediaDecoderError) as exc_info: + await handler._maybe_nvdec_decoder("https://example.com/clip.webm") + + msg = str(exc_info.value) + assert "'vp9'" in msg + assert VALIDATED_SPECS["decord2"] in msg + assert "install_media_decoders sglang" in msg + + +@pytest.mark.asyncio +async def test_vp9_with_software_decoder_passes_bytes_through(monkeypatch): + """With decord importable the preflight stays silent and the validated + bytes are handed to SGLang exactly as before.""" + import dynamo.sglang.request_handlers.multimodal.encode_worker_handler as ewh + + async def fake_validate(url, policy): + return url + + async def fake_fetch(url, timeout, policy=None): + return b"vp9-bytes" + + monkeypatch.setattr(ewh, "validate_media_url", fake_validate) + monkeypatch.setattr(ewh, "fetch_bytes", fake_fetch) + monkeypatch.setattr(ewh, "probe_video_codec", lambda b: "vp9") + monkeypatch.setattr(ewh, "should_use_nvdec", lambda c: False) + monkeypatch.setattr(ewh.importlib, "import_module", _selective_import({"decord"})) + + handler = _bare_handler() + result = await handler._maybe_nvdec_decoder("https://example.com/clip.webm") + + assert result == b"vp9-bytes" diff --git a/components/src/dynamo/trtllm/multimodal_processor.py b/components/src/dynamo/trtllm/multimodal_processor.py index ba4de88c243a..667dc648e028 100644 --- a/components/src/dynamo/trtllm/multimodal_processor.py +++ b/components/src/dynamo/trtllm/multimodal_processor.py @@ -35,7 +35,12 @@ UrlValidationPolicy, validate_media_url, ) +from dynamo.common.multimodal.codec_errors import ( + MissingMediaDecoderError, + video_decoder_missing, +) from dynamo.common.multimodal.image_loader import ImageLoader +from dynamo.common.multimodal.media_source import describe_media_source from dynamo.common.multimodal.nvdec_decoder import probe_video_codec, should_use_nvdec from dynamo.common.multimodal.video_loader import VideoLoader from dynamo.runtime.logging import configure_dynamo_logging @@ -462,13 +467,21 @@ async def process_openai_request( videos = [] for item in video_items: url = item.get("Url") if isinstance(item, dict) else item + # Everything user-supplied that can reach an error message or a + # log line goes through this bounded label: a data: URI carries + # the entire media payload inline, so echoing one back would + # serialize megabytes of base64 to the client and to every log + # sink that records the failure. + source = describe_media_source( + url if isinstance(url, str) else str(item) + ) if not isinstance(url, str): raise HttpStatusError( - 400, f"Unsupported video item: {item!r}", str(item) + 400, f"Unsupported video item: {source}", source ) if urlparse(url).scheme in ("", "file"): raise HttpStatusError( - 400, "Local file access is not allowed for video", url + 400, "Local file access is not allowed for video", source ) try: normalized_url = await validate_media_url(url, self._url_policy) @@ -479,7 +492,8 @@ async def process_openai_request( # Dual decode path: H.264/H.265 via NVDEC (hardware); other # codecs via the vendor cv2 loader. NVDEC failure falls back. nvdec_video = None - if should_use_nvdec(probe_video_codec(content)): + codec = probe_video_codec(content) + if should_use_nvdec(codec): try: nvdec_video = await asyncio.to_thread( _nvdec_video_data, content, self.num_video_frames @@ -498,27 +512,59 @@ async def process_openai_request( ) as video_file: await asyncio.to_thread(video_file.write, content) await asyncio.to_thread(video_file.flush) - videos.append( - await async_load_video( - video_file.name, self.num_video_frames + try: + videos.append( + await async_load_video( + video_file.name, self.num_video_frames + ) ) - ) + except ImportError as exc: + # The vendor loader needs cv2, which the + # image deliberately omits; its bare error + # names neither codec nor remedy. Carry its + # text as the cause so the underlying + # reason still reaches the client. + raise video_decoder_missing( + "trtllm", + "opencv-python-headless", + "cv2", + codec, + cause=str(exc), + ) from exc else: - videos.append( - await async_load_video( - normalized_url, self.num_video_frames + try: + videos.append( + await async_load_video( + normalized_url, self.num_video_frames + ) ) - ) + except ImportError as exc: + # No bytes fetched on this branch, so no codec probe. + raise video_decoder_missing( + "trtllm", + "opencv-python-headless", + "cv2", + None, + cause=str(exc), + ) from exc except UrlValidationError as e: - raise HttpStatusError(400, str(e), url) from e + raise HttpStatusError(400, str(e), source) from e except HttpStatusError: raise + except MissingMediaDecoderError as e: + # A missing decoder is deployment configuration, not a bad + # request: 500, not the 400 the generic handler below + # assigns. The actionable text (codec, bounded spec, + # installer command, vendor cause) is the message. + raise HttpStatusError( + 500, f"Failed to load video ({source}): {e}", source + ) from e except Exception as e: status = getattr(e, "status", None) or getattr(e, "code", None) raise HttpStatusError( status if isinstance(status, int) and status >= 400 else 400, - f"Failed to load video ({url}): {e}", - url, + f"Failed to load video ({source}): {e}", + source, ) from e if videos: processed_mm_data["video"] = videos diff --git a/components/src/dynamo/trtllm/tests/test_trtllm_multimodal_processor.py b/components/src/dynamo/trtllm/tests/test_trtllm_multimodal_processor.py index 0e43cbe0243d..073f7cf4b0ef 100644 --- a/components/src/dynamo/trtllm/tests/test_trtllm_multimodal_processor.py +++ b/components/src/dynamo/trtllm/tests/test_trtllm_multimodal_processor.py @@ -239,8 +239,10 @@ async def test_vp9_video_reports_an_unsupported_codec_error(monkeypatch) -> None ep_disaggregated_params=None, ) - # A client-visible 400, not a 500: the input is unsupported, not broken server. - assert excinfo.value.status == 400 + # Contract change with the actionable-error work: a missing DECODER is + # deployment configuration, not a bad request, so this is now a 500 (the + # ImportError wrap classifies it), not the generic 400. + assert excinfo.value.status == 500 message = str(excinfo.value) # Ours: the prefix and the offending URL, so the client knows which input failed. assert "Failed to load video" in message @@ -248,5 +250,98 @@ async def test_vp9_video_reports_an_unsupported_codec_error(monkeypatch) -> None # The upstream reason is preserved verbatim rather than swallowed. Asserted as # "whatever the cause said survives", not as specific vendor wording. assert str(upstream_error) in message + # And the actionable guidance is present alongside it. + assert "install_media_decoders trtllm" in message nvdec.assert_not_called() # VP9 must never take the hardware path + + +@pytest.mark.asyncio +async def test_video_error_never_echoes_an_inline_media_payload( + monkeypatch, +) -> None: + """A data: video URL carries the whole payload inline, so it must never be + echoed back in the error. + + Measured on the real TRT-LLM image before this bound existed: a 250 KB + inline video produced a 683,213-character message -- the payload twice + (once from our text, once from HttpStatusError's own format string), about + 1500x the 457 characters of actionable guidance -- returned to the client + and written to every log sink that recorded the failure. + """ + import base64 + + payload = base64.b64encode(b"\x00" * (64 * 1024)).decode() + url = f"data:video/mp4;base64,{payload}" + + processor = MultimodalRequestProcessor( + model_type="multimodal", + model_dir="unused", + max_file_size_mb=10, + tokenizer=MagicMock(), + ) + monkeypatch.setattr( + mmp, + "async_load_video", + AsyncMock(side_effect=ImportError("OpenCV (cv2) is required")), + ) + + with pytest.raises(HttpStatusError) as exc_info: + await processor.process_openai_request( + { + "multi_modal_data": {"video_url": [{"Url": url}]}, + "token_ids": [1], + }, + embeddings=None, + ep_disaggregated_params=None, + ) + + message = str(exc_info.value) + assert payload not in message + assert payload not in str(exc_info.value.url) + # Still identifies the source by type and size, and stays actionable. + assert "data:video/mp4" in message + assert "payload elided" in message + assert "install_media_decoders trtllm" in message + assert len(message) < 2_000, f"error message is unbounded: {len(message)} chars" + + +@pytest.mark.asyncio +async def test_video_missing_decoder_error_is_actionable(monkeypatch) -> None: + """cv2 absent: the vendor loader's ImportError must surface as guidance + naming the bounded spec and installer, not a bare module error.""" + from dynamo.common.utils.install_media_decoders import VALIDATED_SPECS + + monkeypatch.setenv("DYN_MM_ALLOW_INTERNAL", "1") + processor = MultimodalRequestProcessor( + model_type="multimodal", + model_dir="unused", + max_file_size_mb=10, + tokenizer=MagicMock(), + ) + fetch = AsyncMock(return_value=b"video bytes") + load_video = AsyncMock( + side_effect=ImportError("OpenCV (cv2) is required for video decoding") + ) + monkeypatch.setattr(mmp, "fetch_bytes", fetch, raising=False) + monkeypatch.setattr(mmp, "async_load_video", load_video) + + with pytest.raises(HttpStatusError) as exc_info: + await processor.process_openai_request( + { + "multi_modal_data": { + "video_url": [{"Url": "http://169.254.169.254/x.mp4"}] + }, + "token_ids": [1], + }, + embeddings=None, + ep_disaggregated_params=None, + ) + + # A missing decoder is a deployment gap: 500, never the generic 400. + assert exc_info.value.status == 500 + msg = str(exc_info.value) + assert VALIDATED_SPECS["opencv-python-headless"] in msg + assert "install_media_decoders trtllm" in msg + # The vendor loader's own text survives as the cause. + assert "OpenCV (cv2) is required for video decoding" in msg diff --git a/docs/fern/pages/developer-guide/knowledge-base/modular-components/backends/sglang/multimodal.md b/docs/fern/pages/developer-guide/knowledge-base/modular-components/backends/sglang/multimodal.md index 174c74b72416..7a2ed0564b67 100644 --- a/docs/fern/pages/developer-guide/knowledge-base/modular-components/backends/sglang/multimodal.md +++ b/docs/fern/pages/developer-guide/knowledge-base/modular-components/backends/sglang/multimodal.md @@ -42,9 +42,15 @@ This document provides a comprehensive guide for multimodal inference using SGLa | Format | Example | Description | |--------|---------|-------------| -| **HTTP/HTTPS** | `http://example.com/image.jpg` | Remote media files | +| **HTTP/HTTPS** | `https://example.com/image.jpg` | Remote media files | | **file://** | `file:///tmp/test.mp4` | Local files accessible to the backend | +> [!NOTE] +> Media URLs are validated against a default-deny policy. `https://` and `data:` sources +> pass; plain `http://` and hostnames that resolve to private or loopback addresses are +> refused. To fetch media over the cluster's internal network, set +> `DYN_MM_ALLOW_INTERNAL=1` on the worker that loads it. + ## Deployment Patterns SGLang supports EPD, EP/D, E/PD, and E/P/D patterns. See [Multimodal Model Serving](../../../../../use-cases/multimodal-serving/overview.md) for detailed explanations. diff --git a/docs/fern/pages/developer-guide/knowledge-base/modular-components/backends/tensorrt-llm/multimodal.md b/docs/fern/pages/developer-guide/knowledge-base/modular-components/backends/tensorrt-llm/multimodal.md index dfb4ddbc372b..ac1354ce6d0a 100644 --- a/docs/fern/pages/developer-guide/knowledge-base/modular-components/backends/tensorrt-llm/multimodal.md +++ b/docs/fern/pages/developer-guide/knowledge-base/modular-components/backends/tensorrt-llm/multimodal.md @@ -25,9 +25,15 @@ You can provide multimodal inputs in the following ways: | Format | Example | Description | |--------|---------|-------------| -| **HTTP/HTTPS** | `http://example.com/image.jpg` | Remote media files | +| **HTTP/HTTPS** | `https://example.com/image.jpg` | Remote media files | | **Pre-computed Embeddings** | `/path/to/embedding.safetensors` | Local embedding files (.safetensors only) | +> [!NOTE] +> Media URLs are validated against a default-deny policy. `https://` and `data:` sources +> pass; plain `http://` and hostnames that resolve to private or loopback addresses are +> refused. To fetch media over the cluster's internal network, set +> `DYN_MM_ALLOW_INTERNAL=1` on the worker that loads it. + ## Deployment Patterns TRT-LLM supports aggregated and traditional disaggregated patterns. See [Multimodal Model Serving](../../../../../use-cases/multimodal-serving/overview.md) for detailed explanations. diff --git a/docs/fern/pages/developer-guide/knowledge-base/modular-components/backends/vllm/multimodal.md b/docs/fern/pages/developer-guide/knowledge-base/modular-components/backends/vllm/multimodal.md index dcc6d44ceeaf..520f1c53a88a 100644 --- a/docs/fern/pages/developer-guide/knowledge-base/modular-components/backends/vllm/multimodal.md +++ b/docs/fern/pages/developer-guide/knowledge-base/modular-components/backends/vllm/multimodal.md @@ -36,9 +36,15 @@ This document provides a comprehensive guide for multimodal inference using the | Format | Example | Description | | -------------- | ------------------------------------ | -------------------------- | -| **HTTP/HTTPS** | `http://example.com/image.jpg` | Remote media files | +| **HTTP/HTTPS** | `https://example.com/image.jpg` | Remote media files | | **Data URL** | `data:image/jpeg;base64,/9j/4AAQ...` | Base64-encoded inline data | +> [!NOTE] +> Media URLs are validated against a default-deny policy. `https://` and `data:` sources +> pass; plain `http://` and hostnames that resolve to private or loopback addresses are +> refused. To fetch media over the cluster's internal network, set +> `DYN_MM_ALLOW_INTERNAL=1` on the worker that loads it. + ## Deployment Patterns The main multimodal vLLM launchers in this repo are: diff --git a/lib/llm/src/preprocessor/media/decoders/video.rs b/lib/llm/src/preprocessor/media/decoders/video.rs index dd13ede877fa..36a5e3ace4a3 100644 --- a/lib/llm/src/preprocessor/media/decoders/video.rs +++ b/lib/llm/src/preprocessor/media/decoders/video.rs @@ -228,7 +228,22 @@ impl Decoder for VideoDecoder { mem_file.add_seals(Seal::Write | Seal::Shrink | Seal::Grow)?; let fd_path = format!("/proc/self/fd/{}", mem_file.as_raw_fd()); let location = Location::File(fd_path.into()); - let mut decoder = video_rs::decode::Decoder::new(location)?; + // `Decoder::new` can fail for an unsupported codec (the in-tree FFmpeg + // decodes only VP8/VP9 -- H.264/H.265/etc. are not built in) but also + // for malformed / non-video input. Keep the original FFmpeg error + // prominent and frame the codec guidance conditionally so a bad + // payload is not misreported as a codec problem. + let mut decoder = video_rs::decode::Decoder::new(location).map_err(|e| { + anyhow::anyhow!( + "failed to open the video for decoding: {e}. If the input uses a \ + codec other than VP8/VP9 (e.g. H.264 or H.265), note this \ + frontend decoder's in-tree FFmpeg decodes only VP8/VP9 -- \ + re-encode to VP9, e.g. `ffmpeg -i input.mp4 -c:v libvpx-vp9 -an \ + output.webm`, or send it to the backend, where H.264/H.265 \ + decode in hardware via NVDEC. Otherwise the input may be \ + malformed or not a video." + ) + })?; let requested_frames = get_num_requested_frames(self, &decoder)?; let source_duration = decoder.duration()?.as_secs() as f64; @@ -346,6 +361,34 @@ mod tests { (encoded, width, height, frames) } + #[test] + fn test_unsupported_codec_error_is_actionable() { + // H.264 fixture: the in-tree FFmpeg decodes only VP8/VP9, so opening + // must fail -- and with the re-encode guidance, not ffmpeg's bare + // "Decoder not found". + let path = format!( + "{}/tests/data/media/triangle_240p_10_h264.mp4", + env!("CARGO_MANIFEST_DIR") + ); + let bytes = std::fs::read(&path).expect("h264 fixture must exist"); + let encoded = EncodedMediaData { + bytes, + b64_encoded: false, + }; + let decoder = VideoDecoder { + limits: VideoDecoderLimits::default(), + fps: None, + max_frames: None, + num_frames: Some(2), + strict: false, + }; + + let err = decoder.decode(encoded).expect_err("h264 must not decode"); + let msg = format!("{err:#}"); + assert!(msg.contains("VP8/VP9"), "no codec guidance in: {msg}"); + assert!(msg.contains("libvpx-vp9"), "no re-encode hint in: {msg}"); + } + #[test] fn test_decode_video_num_frames() { let (encoded_data, width, height, _total_frames) = load_test_video("240p_10.mp4"); diff --git a/tests/serve/test_sglang.py b/tests/serve/test_sglang.py index d7b942aca048..46f82ae9e50f 100644 --- a/tests/serve/test_sglang.py +++ b/tests/serve/test_sglang.py @@ -719,6 +719,13 @@ class SGLangConfig(EngineConfig): "DYN_ENCODE_GPU_MEM": "0.1", "DYN_WORKER_GPU_MEM": "0.4", "DYN_SGL_EMBEDDING_TRANSFER_MODE": "local", + # The clips come from the image_server over plain http on localhost, + # which the URL policy rejects by default. This model is gated out of + # NVDEC (see _NVDEC_UNSAFE_MODEL_TYPES), and that disabled path now + # applies the policy just like the NVDEC path already did -- so this + # opt-in is required here for the same reason as in the nvdec config + # above, not because the two tests differ. + "DYN_MM_ALLOW_INTERNAL": "1", # SGLang's video path decodes with decord; the shipped image omits it # as a media-codec carrier, so install it for this test only. "DYN_TEST_ONLY_PIP_INSTALL": VALIDATED_SPECS["decord2"],