fix(apps): accept datacenter ids outside the built-in list

runpod/apps/datacenter.py

@@ -40,6 +40,22 @@ def from_string(cls, value: str) -> "DataCenter":
     def all(cls) -> List["DataCenter"]:
         return list(cls)
 
+    @classmethod
+    def is_known(cls, datacenter_id: str) -> bool:
+        return datacenter_id in cls._value2member_map_
+
+
+def datacenter_id(value: str) -> str:
+    """api id for a datacenter: built-in ids normalize, others pass through.
+
+    the built-in list goes stale as datacenters come online, and the
+    backend validates ids anyway, so unrecognized ids are only stripped.
+    """
+    try:
+        return DataCenter.from_string(value).value
+    except ValueError:
+        return value.strip()
+
 
 # datacenters with high cpu serverless stock. storage support is checked
 # separately; cpu5c/cpu5g are only stocked in EU-RO-1.

runpod/apps/placement.py

@@ -11,7 +11,7 @@
 import logging
 from typing import Dict, Iterable, List, Optional, Set, Tuple
 
-from .datacenter import DataCenter
+from .datacenter import DataCenter, datacenter_id
 from .errors import AppError
 from .gpu import GpuGroup
 from .utils.client import default_client
@@ -161,17 +161,21 @@ def candidates(spec, stock: StockMap) -> Set[str]:
     """
     spec.validate()
     allowed = {dc.value for dc in DataCenter.all()}
+    # stock is only queried for built-in datacenters, so an unlisted pin
+    # cannot be ruled out here; the backend decides whether it schedules
+    unlisted: Set[str] = set()
     if spec.datacenter:
+        unlisted = {dc for dc in spec.datacenter if not DataCenter.is_known(dc)}
         allowed &= set(spec.datacenter)
 
     keys = _hardware_keys(spec)
     if not keys:
-        return allowed
+        return allowed | unlisted
     viable = set()
     for dc in allowed:
         if any(stock.score(key, dc) > 0 for key in keys):
             viable.add(dc)
-    return viable
+    return viable | unlisted
 
 
 def _resource_best_score(spec, stock: StockMap, dc: str) -> int:
@@ -200,7 +204,7 @@ def solve_placement(
     per_resource = {spec.name: candidates(spec, stock) for spec in specs}
 
     if existing_dc is not None:
-        existing_dc = DataCenter.from_string(existing_dc).value
+        existing_dc = datacenter_id(existing_dc)
         blocked = [name for name, dcs in per_resource.items() if existing_dc not in dcs]
         if blocked:
             raise PlacementError(

runpod/apps/spec.py

@@ -5,17 +5,25 @@
 and what dev-session provisioning consumes. it holds no live state.
 """
 
+import logging
 from dataclasses import dataclass, field
 from enum import Enum
+from functools import lru_cache
 from typing import Any, Dict, List, Mapping, Optional, Tuple, Union
 
 from .errors import InvalidResourceError
-from .datacenter import DataCenter
+from .datacenter import DataCenter, datacenter_id
 from .gpu import GpuGroup, GpuLike, GpuType
 from .volume import Volume, normalize_mounts, validate_mounts
 
+log = logging.getLogger(__name__)
+
 DEFAULT_WORKERS: Tuple[int, int] = (0, 3)
 
+DATACENTER_IDS_DOCS_URL = (
+    "https://docs.runpod.io/api-reference-v2/catalog/list-data-centers"
+)
+
 SCALER_TYPES = frozenset({"QUEUE_DELAY", "REQUEST_COUNT"})
 DEFAULT_SCALER_VALUE = 4
 
@@ -147,17 +155,37 @@ def normalize_cuda_version(version: Optional[str]) -> Optional[str]:
     return normalized
 
 
+@lru_cache(maxsize=None)
+def _warn_unlisted_datacenter(dc_id: str) -> None:
+    # specs revalidate many times per deploy; warn once per id per process
+    log.warning(
+        "%r doesn't look like a valid datacenter id (see %s) but will be used anyway",
+        dc_id,
+        DATACENTER_IDS_DOCS_URL,
+    )
+
+
 def normalize_datacenter(
     datacenter: Optional[Union[str, List[str]]],
 ) -> Optional[List[str]]:
-    """normalize enum values and string variants to api datacenter ids."""
+    """normalize enum values and string variants to api datacenter ids.
+
+    ids outside the built-in DataCenter list pass through with a warning;
+    blank or non-string values raise InvalidResourceError.
+    """
     if datacenter is None:
         return None
     values = [datacenter] if isinstance(datacenter, str) else datacenter
     try:
-        return [DataCenter.from_string(value).value for value in values]
-    except (AttributeError, TypeError, ValueError) as exc:
+        ids = [datacenter_id(value) for value in values]
+    except (AttributeError, TypeError) as exc:
         raise InvalidResourceError(f"invalid datacenter: {datacenter!r}") from exc
+    if not all(ids):
+        raise InvalidResourceError(f"invalid datacenter: {datacenter!r}")
+    for dc_id in ids:
+        if not DataCenter.is_known(dc_id):
+            _warn_unlisted_datacenter(dc_id)
+    return ids
 
 
 @dataclass

runpod/apps/volume.py

@@ -335,10 +335,10 @@ async def resolve(
 
         dc = record["dataCenter"] if record is not None else volume.datacenter
         if record is None:
-            from .datacenter import DataCenter
+            from .datacenter import datacenter_id
 
             pins = {
-                DataCenter.from_string(ref.datacenter).value
+                datacenter_id(ref.datacenter)
                 for ref in [volume]
                 + [
                     ref

tests/test_apps/test_config_params.py

@@ -4,15 +4,20 @@
 
 from runpod.apps import App
 from runpod.apps.app import _clear_registry
+from runpod.apps.datacenter import DataCenter
 from runpod.apps.dev import _endpoint_input
 from runpod.apps.errors import InvalidResourceError
 from runpod.apps.spec import (
+    DATACENTER_IDS_DOCS_URL,
     ResourceKind,
     ResourceSpec,
     normalize_cuda_version,
+    normalize_datacenter,
     normalize_scaler_type,
 )
 
+UNLISTED_DC = "XX-TEST-1"
+
 
 @pytest.fixture(autouse=True)
 def clean_registry():
@@ -44,6 +49,55 @@ def test_cuda_version_rejects_unknown(self):
             normalize_cuda_version("10.0")
 
 
+class TestDatacenterNormalization:
+    @pytest.fixture(autouse=True)
+    def _fresh_warnings(self):
+        from runpod.apps.spec import _warn_unlisted_datacenter
+
+        _warn_unlisted_datacenter.cache_clear()
+        yield
+        _warn_unlisted_datacenter.cache_clear()
+
+    @pytest.mark.parametrize(
+        "value", [DataCenter.EU_RO_1, "EU-RO-1", "eu_ro_1", " eu-ro-1 ", ["eu_ro_1"]]
+    )
+    def test_known_values_normalize_without_warning(self, value, caplog):
+        assert normalize_datacenter(value) == ["EU-RO-1"]
+        assert "valid datacenter id" not in caplog.text
+
+    def test_none_passes(self):
+        assert normalize_datacenter(None) is None
+
+    def test_unlisted_id_passes_through_with_warning(self, caplog):
+        assert normalize_datacenter(UNLISTED_DC) == [UNLISTED_DC]
+        warnings = [r for r in caplog.records if r.levelname == "WARNING"]
+        assert len(warnings) == 1
+        assert UNLISTED_DC in warnings[0].getMessage()
+        message = warnings[0].getMessage()
+        assert "doesn't look like a valid datacenter id" in message
+        assert "will be used anyway" in message
+        assert DATACENTER_IDS_DOCS_URL in message
+
+    def test_mixed_list_keeps_order(self):
+        assert normalize_datacenter(["us_ks_2", UNLISTED_DC]) == [
+            "US-KS-2",
+            UNLISTED_DC,
+        ]
+
+    def test_unlisted_id_warns_once_across_revalidation(self, caplog):
+        spec = ResourceSpec(kind=ResourceKind.QUEUE, name="q", datacenter=UNLISTED_DC)
+        spec.validate()
+        spec.to_manifest()
+        warnings = [r for r in caplog.records if UNLISTED_DC in r.getMessage()]
+        assert len(warnings) == 1
+        assert spec.datacenter == [UNLISTED_DC]
+
+    @pytest.mark.parametrize("value", ["", "   ", [""], ["EU-RO-1", " "], [None], 42])
+    def test_blank_or_non_string_rejected(self, value):
+        with pytest.raises(InvalidResourceError, match="invalid datacenter"):
+            normalize_datacenter(value)
+
+
 class TestSpecValidation:
     @pytest.mark.parametrize(
         "config",

tests/test_apps/test_deploy.py

@@ -132,6 +132,20 @@ def value(self):
         assert env["PORT_HEALTH"] == "80"
 
 
+    @pytest.mark.parametrize(
+        "hardware", [{"gpu": "ADA_24"}, {"cpu": "cpu3c-1-2"}, {"cpu": "cpu5c-2-4"}]
+    )
+    def test_unlisted_datacenter_reaches_locations(self, hardware):
+        app = App("dc-app")
+
+        @app.queue(name="q", datacenter="XX-TEST-1", **hardware)
+        def q():
+            pass
+
+        payload = _deployed_endpoint_input(app, q.spec, "env-1", "build-1", "3.12")
+        assert payload["locations"] == "XX-TEST-1"
+
+
 class TestPackaging:
     def test_tarball_contains_source_and_manifest(self, tmp_path):
         _write_project(tmp_path)

tests/test_apps/test_placement.py

@@ -79,6 +79,20 @@ def test_datacenter_variants_match_api_ids(self, pin):
         assert candidates(spec, stock) == {"EU-RO-1"}
         assert spec.to_manifest()["locations"] == "EU-RO-1"
 
+    def test_unlisted_pin_stays_candidate_without_stock_data(self):
+        # the sdk only queries stock for its built-in list, so an unlisted
+        # pin cannot be ruled out client-side; the backend decides
+        stock = _stock(gpu={("NVIDIA GeForce RTX 4090", "EU-RO-1"): 3})
+        spec = _spec(gpu=["NVIDIA GeForce RTX 4090"], datacenter=["XX-TEST-1"])
+        assert candidates(spec, stock) == {"XX-TEST-1"}
+
+    def test_unlisted_pin_alongside_known_pin(self):
+        stock = _stock(gpu={("NVIDIA GeForce RTX 4090", "EU-RO-1"): 0})
+        spec = _spec(
+            gpu=["NVIDIA GeForce RTX 4090"], datacenter=["EU-RO-1", "XX-TEST-1"]
+        )
+        assert candidates(spec, stock) == {"XX-TEST-1"}
+
     def test_mutated_datacenter_pin_is_normalized(self):
         spec = _spec(cpu="cpu5c-2-4")
         spec.datacenter = [DataCenter.EU_RO_1]
@@ -145,6 +159,36 @@ def test_existing_dc_is_hard_constraint(self):
         )
         assert dc == "EU-RO-1"
 
+    def test_existing_unlisted_dc_with_matching_pin(self):
+        dc = solve_placement(
+            [_spec("train", gpu=["NVIDIA H200"], datacenter="XX-TEST-1")],
+            _stock(),
+            volume_name="models",
+            existing_dc="XX-TEST-1",
+            volume_datacenters=set(),
+        )
+        assert dc == "XX-TEST-1"
+
+    def test_existing_unlisted_dc_without_pin_is_placement_error(self):
+        with pytest.raises(PlacementError, match="lives in XX-TEST-1"):
+            solve_placement(
+                [_spec("train", gpu=["NVIDIA H200"])],
+                _stock(gpu={("NVIDIA H200", "EU-RO-1"): 3}),
+                volume_name="models",
+                existing_dc="XX-TEST-1",
+                volume_datacenters=set(),
+            )
+
+    def test_new_volume_in_unlisted_dc(self):
+        dc = solve_placement(
+            [_spec("train", gpu=["NVIDIA H200"], datacenter="XX-TEST-1")],
+            _stock(),
+            volume_name="models",
+            volume_dc="XX-TEST-1",
+            volume_datacenters={"EU-RO-1", "XX-TEST-1"},
+        )
+        assert dc == "XX-TEST-1"
+
     def test_existing_dc_unschedulable_errors(self):
         stock = _stock(gpu={("NVIDIA H200", "EU-RO-1"): 3})
         with pytest.raises(PlacementError, match="lives in US-KS-2"):

tests/test_apps/test_volume.py

@@ -317,6 +317,22 @@ def test_creation_without_app_requires_explicit_placement(self):
         )
         assert resolved == {"id": "nv-new", "dataCenterId": "EU-RO-1"}
 
+    async def test_unlisted_volume_pin_uses_catalog(self):
+        api = _api(
+            created={"id": "nv-new", "name": "models", "size": 50, "dataCenter": "XX-TEST-1"}
+        )
+        api.network_volume_datacenters.return_value = {"EU-RO-1", "XX-TEST-1"}
+        spec = ResourceSpec(
+            kind=ResourceKind.TASK, name="r", gpu=None, datacenter="XX-TEST-1"
+        )
+        resolved = await VolumeResolver(api).resolve(
+            NetworkVolume("models", datacenter="XX-TEST-1"), [spec]
+        )
+        assert resolved == {"id": "nv-new", "dataCenterId": "XX-TEST-1"}
+        api.create_network_volume.assert_awaited_once_with(
+            name="models", size=50, data_center_id="XX-TEST-1"
+        )
+
     async def test_unsupported_explicit_pin_never_relocates(self):
         api = _api()
         api.network_volume_datacenters.return_value = {"EU-RO-1"}