feat(sglang): upload sidecar metadata with OpenDAL

Cargo.toml

@@ -132,6 +132,13 @@ ipnet = { version = "2" }
 libc = { version = "0.2" }
 oneshot = { version = "0.1.13", features = ["std", "async"] }
 ordered-float = "4"
+opendal = { version = "0.59.1", default-features = false, features = [
+    "http-transport-reqwest",
+    "services-azblob",
+    "services-fs",
+    "services-gcs",
+    "services-s3",
+] }
 parking_lot = "0.12.5"
 prometheus = { version = "0.14"}
 rand = { version = "0.9.2" }
@@ -188,6 +195,7 @@ uuid = { version = "1.18.1", features = ["v4", "serde"] }
 url = { version = "2.5", features = ["serde"] }
 xxhash-rust = { version = "0.8", features = ["xxh3", "const_xxh3"] }
 zeroize = { version = "1" }
+zstd = "0.13"
 
 # ext-proc / Envoy gRPC
 prost = { version = "0.13.5" }

components/src/dynamo/common/metadata_upload.py

@@ -4,11 +4,15 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 from dataclasses import dataclass
 from functools import cache
 from typing import Any, cast
 
 
+logger = logging.getLogger(__name__)
+
+
 def _backend_metadata_upload_settings(request: dict[str, Any]) -> dict[str, Any] | None:
     nvext = request.get("nvext")
     extra_args = request.get("extra_args") or {}
@@ -138,29 +142,47 @@ def _normalize_for_upload(value: Any) -> Any:
     return value
 
 
+def _normalize_metadata_upload_url(value: object, field: str) -> str:
+    if not isinstance(value, str):
+        raise ValueError(f"metadata_upload.{field} must be a string")
+    value = value.strip()
+    if not value:
+        raise ValueError(f"metadata_upload.{field} must not be empty")
+    return value
+
+
 @dataclass(frozen=True)
 class MetadataUploader:
     url: str
+    fallback_url: str | None = None
 
     def __post_init__(self) -> None:
-        if not isinstance(self.url, str):
-            raise ValueError("metadata_upload.url must be a string")
-        url = self.url.strip()
-        if not url:
-            raise ValueError("metadata_upload.url must not be empty")
-        object.__setattr__(self, "url", url)
+        object.__setattr__(
+            self,
+            "url",
+            _normalize_metadata_upload_url(self.url, "url"),
+        )
+        if self.fallback_url is not None:
+            object.__setattr__(
+                self,
+                "fallback_url",
+                _normalize_metadata_upload_url(self.fallback_url, "fallback_url"),
+            )
 
     @classmethod
     def from_settings(cls, settings: dict[str, Any] | None) -> MetadataUploader | None:
         if settings is None:
             return None
-        unexpected = settings.keys() - {"url"}
+        unexpected = settings.keys() - {"url", "fallback_url"}
         if unexpected:
             field = min(unexpected)
             raise ValueError(f"metadata_upload.{field} is not supported")
         if "url" not in settings:
             raise ValueError("metadata_upload.url is required")
-        return cls(url=settings["url"])
+        return cls(
+            url=settings["url"],
+            fallback_url=settings.get("fallback_url"),
+        )
 
     @classmethod
     def from_backend_request(cls, request: dict[str, Any]) -> MetadataUploader | None:
@@ -173,6 +195,15 @@ async def upload_choice(self, choice_index: int, metadata: dict[str, Any]) -> No
         storage_path = f"choice_{choice_index}.msgpack.zst"
         data = await asyncio.to_thread(_serialize_metadata, metadata)
         try:
-            await _upload_bytes(self.url, storage_path, data)
+            try:
+                await _upload_bytes(self.url, storage_path, data)
+            except Exception:
+                if self.fallback_url is None:
+                    raise
+                logger.warning(
+                    "Primary metadata upload failed; attempting configured fallback",
+                    exc_info=True,
+                )
+                await _upload_bytes(self.fallback_url, storage_path, data)
         finally:
             del data

components/src/dynamo/common/tests/test_metadata_upload.py

@@ -0,0 +1,167 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Unit tests for out-of-band metadata uploads."""
+
+import msgspec
+import pytest
+import zstandard as zstd
+
+from dynamo.common.metadata_upload import MetadataUploader
+
+pytestmark = [
+    pytest.mark.unit,
+    pytest.mark.gpu_0,
+    pytest.mark.pre_merge,
+]
+
+
+def _read_zstd_payload(path):
+    raw = zstd.ZstdDecompressor().decompress(path.read_bytes())
+    return msgspec.msgpack.decode(raw)
+
+
+def test_metadata_uploader_parses_backend_request():
+    uploader = MetadataUploader.from_backend_request(
+        {
+            "extra_args": {
+                "nvext": {
+                    "metadata_upload": {
+                        "url": "s3://bucket/root/rollouts",
+                        "fallback_url": " file:///var/tmp/rollouts ",
+                    }
+                }
+            }
+        }
+    )
+
+    assert uploader is not None
+    assert uploader.url == "s3://bucket/root/rollouts"
+    assert uploader.fallback_url == "file:///var/tmp/rollouts"
+
+    uploader = MetadataUploader.from_backend_request(
+        {"nvext": {"metadata_upload": {"url": "s3://bucket/root/rollouts"}}}
+    )
+    assert uploader is not None
+    assert uploader.fallback_url is None
+
+
+@pytest.mark.parametrize(
+    ("settings", "message"),
+    [
+        ({}, "metadata_upload.url is required"),
+        ({"url": ""}, "metadata_upload.url must not be empty"),
+        ({"url": 1}, "metadata_upload.url must be a string"),
+        (
+            {"url": "s3://bucket/root/rollouts", "fallback_url": ""},
+            "metadata_upload.fallback_url must not be empty",
+        ),
+        (
+            {"url": "s3://bucket/root/rollouts", "fallback_url": 1},
+            "metadata_upload.fallback_url must be a string",
+        ),
+        ("invalid", "metadata_upload must be an object"),
+        (
+            {"url": "s3://bucket/root/rollouts", "format": "json"},
+            "metadata_upload.format is not supported",
+        ),
+    ],
+)
+def test_metadata_uploader_rejects_invalid_settings(settings, message):
+    with pytest.raises(ValueError, match=message):
+        MetadataUploader.from_backend_request({"nvext": {"metadata_upload": settings}})
+
+
+@pytest.mark.asyncio
+async def test_metadata_upload_normalizes_tensor_values(tmp_path):
+    torch = pytest.importorskip("torch")
+    uploader = MetadataUploader(url=(tmp_path / "metadata/tensors").as_uri())
+    tensor = torch.tensor([[1, 2], [3, 4]], dtype=torch.int32)
+
+    await uploader.upload_choice(0, {"router_tensor": tensor})
+
+    payload = _read_zstd_payload(tmp_path / "metadata/tensors/choice_0.msgpack.zst")
+    uploaded_tensor = payload["metadata"]["router_tensor"]
+    assert uploaded_tensor["type"] == "tensor"
+    assert uploaded_tensor["dtype"] == "int32"
+    assert uploaded_tensor["shape"] == [2, 2]
+    assert uploaded_tensor["data"] == tensor.numpy().tobytes()
+
+    bfloat_tensor = torch.tensor([1, 2], dtype=torch.bfloat16)
+    await uploader.upload_choice(1, {"router_tensor": bfloat_tensor})
+
+    payload = _read_zstd_payload(tmp_path / "metadata/tensors/choice_1.msgpack.zst")
+    uploaded_tensor = payload["metadata"]["router_tensor"]
+    assert uploaded_tensor["dtype"] == "bfloat16"
+    assert uploaded_tensor["shape"] == [2]
+    assert uploaded_tensor["data"] == bfloat_tensor.view(torch.uint8).numpy().tobytes()
+
+
+@pytest.mark.asyncio
+async def test_metadata_upload_normalizes_numpy_values(tmp_path):
+    np = pytest.importorskip("numpy")
+    uploader = MetadataUploader(url=(tmp_path / "metadata/arrays").as_uri())
+    array = np.array([[1.5, 2.5], [3.5, 4.5]], dtype=np.float32)
+
+    await uploader.upload_choice(0, {"scores": array[:, ::-1]})
+
+    payload = _read_zstd_payload(tmp_path / "metadata/arrays/choice_0.msgpack.zst")
+    uploaded_array = payload["metadata"]["scores"]
+    expected = np.ascontiguousarray(array[:, ::-1])
+    assert uploaded_array["type"] == "ndarray"
+    assert uploaded_array["dtype"] == "float32"
+    assert uploaded_array["shape"] == [2, 2]
+    assert uploaded_array["data"] == expected.tobytes()
+
+
+@pytest.mark.asyncio
+async def test_metadata_upload_does_not_write_fallback_when_primary_succeeds(tmp_path):
+    primary_path = tmp_path / "metadata/primary"
+    fallback_path = tmp_path / "metadata/fallback"
+    uploader = MetadataUploader(
+        url=primary_path.as_uri(),
+        fallback_url=fallback_path.as_uri(),
+    )
+
+    await uploader.upload_choice(0, {"id": "request-1"})
+
+    assert (primary_path / "choice_0.msgpack.zst").is_file()
+    assert not fallback_path.exists()
+
+
+@pytest.mark.asyncio
+async def test_metadata_upload_uses_fallback_after_primary_failure(tmp_path, caplog):
+    primary_path = tmp_path / "primary-is-a-file"
+    primary_path.write_bytes(b"primary sentinel")
+    fallback_path = tmp_path / "metadata/fallback"
+    uploader = MetadataUploader(
+        url=primary_path.as_uri(),
+        fallback_url=fallback_path.as_uri(),
+    )
+
+    with caplog.at_level("WARNING"):
+        await uploader.upload_choice(2, {"id": "request-2"})
+
+    assert primary_path.read_bytes() == b"primary sentinel"
+    assert "Primary metadata upload failed" in caplog.text
+    payload = _read_zstd_payload(fallback_path / "choice_2.msgpack.zst")
+    assert payload["metadata"] == {"id": "request-2"}
+
+
+@pytest.mark.asyncio
+async def test_metadata_upload_propagates_fallback_failure(tmp_path):
+    primary_path = tmp_path / "primary-is-a-file"
+    fallback_path = tmp_path / "fallback-is-a-file"
+    primary_path.write_bytes(b"primary sentinel")
+    fallback_path.write_bytes(b"fallback sentinel")
+    uploader = MetadataUploader(
+        url=primary_path.as_uri(),
+        fallback_url=fallback_path.as_uri(),
+    )
+
+    with pytest.raises(OSError) as exc_info:
+        await uploader.upload_choice(0, {"id": "request-3"})
+
+    assert str(fallback_path) in str(exc_info.value)
+    assert primary_path.read_bytes() == b"primary sentinel"
+    assert fallback_path.read_bytes() == b"fallback sentinel"

components/src/dynamo/frontend/tests/test_sglang_processor_unit.py

@@ -530,6 +530,7 @@ def test_metadata_upload_nvext_is_forwarded_to_backend(self):
                 "nvext": {
                     "metadata_upload": {
                         "url": "s3://bucket/root/rollouts",
+                        "fallback_url": "file:///var/tmp/rollouts",
                     },
                 },
             },
@@ -540,6 +541,7 @@ def test_metadata_upload_nvext_is_forwarded_to_backend(self):
 
         assert result["extra_args"]["nvext"]["metadata_upload"] == {
             "url": "s3://bucket/root/rollouts",
+            "fallback_url": "file:///var/tmp/rollouts",
         }
 
     def test_model_name_and_token_ids(self):

components/src/dynamo/sglang/tests/test_sglang_decode_handler.py

@@ -1061,46 +1061,6 @@ def test_extract_logprobs_formats_top_tokens_as_token_ids():
     ]
 
 
-def test_metadata_uploader_parses_extra_args_nvext():
-    uploader = MetadataUploader.from_backend_request(
-        {
-            "extra_args": {
-                "nvext": {
-                    "metadata_upload": {
-                        "url": "s3://bucket/root/rollouts",
-                    }
-                }
-            }
-        }
-    )
-
-    assert uploader is not None
-    assert uploader.url == "s3://bucket/root/rollouts"
-
-    uploader = MetadataUploader.from_backend_request(
-        {"nvext": {"metadata_upload": {"url": "s3://bucket/root/rollouts"}}}
-    )
-    assert uploader is not None
-
-
-@pytest.mark.parametrize(
-    ("settings", "message"),
-    [
-        ({}, "metadata_upload.url is required"),
-        ({"url": ""}, "metadata_upload.url must not be empty"),
-        ({"url": 1}, "metadata_upload.url must be a string"),
-        ("invalid", "metadata_upload must be an object"),
-        (
-            {"url": "s3://bucket/root/rollouts", "format": "json"},
-            "metadata_upload.format is not supported",
-        ),
-    ],
-)
-def test_metadata_uploader_rejects_invalid_settings(settings, message):
-    with pytest.raises(ValueError, match=message):
-        MetadataUploader.from_backend_request({"nvext": {"metadata_upload": settings}})
-
-
 def test_nvext_extra_field_requested_supports_raw_and_preprocessed_shapes():
     assert _nvext_extra_field_requested(
         {"nvext": {"extra_fields": ["stop_reason"]}}, "stop_reason"
@@ -1131,48 +1091,6 @@ def test_metadata_upload_requires_enable_rl():
     assert uploader.url == "s3://bucket/root/rollouts/run-1"
 
 
-@pytest.mark.asyncio
-async def test_metadata_upload_normalizes_tensor_values(tmp_path):
-    torch = pytest.importorskip("torch")
-    uploader = MetadataUploader(url=(tmp_path / "metadata/tensors").as_uri())
-    tensor = torch.tensor([[1, 2], [3, 4]], dtype=torch.int32)
-
-    await uploader.upload_choice(0, {"router_tensor": tensor})
-
-    payload = _read_zstd_payload(tmp_path / "metadata/tensors/choice_0.msgpack.zst")
-    uploaded_tensor = payload["metadata"]["router_tensor"]
-    assert uploaded_tensor["type"] == "tensor"
-    assert uploaded_tensor["dtype"] == "int32"
-    assert uploaded_tensor["shape"] == [2, 2]
-    assert uploaded_tensor["data"] == tensor.numpy().tobytes()
-
-    bfloat_tensor = torch.tensor([1, 2], dtype=torch.bfloat16)
-    await uploader.upload_choice(1, {"router_tensor": bfloat_tensor})
-
-    payload = _read_zstd_payload(tmp_path / "metadata/tensors/choice_1.msgpack.zst")
-    uploaded_tensor = payload["metadata"]["router_tensor"]
-    assert uploaded_tensor["dtype"] == "bfloat16"
-    assert uploaded_tensor["shape"] == [2]
-    assert uploaded_tensor["data"] == bfloat_tensor.view(torch.uint8).numpy().tobytes()
-
-
-@pytest.mark.asyncio
-async def test_metadata_upload_normalizes_numpy_values(tmp_path):
-    np = pytest.importorskip("numpy")
-    uploader = MetadataUploader(url=(tmp_path / "metadata/arrays").as_uri())
-    array = np.array([[1.5, 2.5], [3.5, 4.5]], dtype=np.float32)
-
-    await uploader.upload_choice(0, {"scores": array[:, ::-1]})
-
-    payload = _read_zstd_payload(tmp_path / "metadata/arrays/choice_0.msgpack.zst")
-    uploaded_array = payload["metadata"]["scores"]
-    expected = np.ascontiguousarray(array[:, ::-1])
-    assert uploaded_array["type"] == "ndarray"
-    assert uploaded_array["dtype"] == "float32"
-    assert uploaded_array["shape"] == [2, 2]
-    assert uploaded_array["data"] == expected.tobytes()
-
-
 @pytest.mark.asyncio
 async def test_process_token_stream_treats_completion_usage_as_optional():
     handler = _new_decode_handler()