OCI: resolve availability domains from the configured compartment

src/dstack/_internal/core/backends/oci/region.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from functools import cached_property
+from functools import cached_property, lru_cache
 from typing import Any, Dict, Iterable, List, Mapping, Set
 
 import oci
@@ -46,7 +46,21 @@ def work_request_client(self) -> oci.work_requests.WorkRequestClient:
 
     @cached_property
     def availability_domains(self) -> List[oci.identity.models.AvailabilityDomain]:
-        return self.identity_client.list_availability_domains(self.client_config["tenancy"]).data
+        return self.availability_domains_in(self.client_config["tenancy"])
+
+    @lru_cache(maxsize=None)
+    def availability_domains_in(
+        self, compartment_id: str
+    ) -> List[oci.identity.models.AvailabilityDomain]:
+        """
+        Availability domains of the tenancy that owns `compartment_id`.
+
+        The compartment dstack works in need not belong to the tenancy the credentials
+        authenticate against - a compartment can be shared with another tenancy by a
+        cross-tenancy policy. Availability domain names are tenancy-specific, so they
+        have to be resolved from the compartment rather than from the credentials.
+        """
+        return self.identity_client.list_availability_domains(compartment_id).data
 
 
 def make_region_client(region_name: str, creds: AnyOCICreds) -> OCIRegionClient:

src/dstack/_internal/core/backends/oci/resources.py

@@ -140,7 +140,7 @@ def list_shapes_in_region(region: OCIRegionClient, compartment_id: str) -> Dict[
     """
 
     result = {}
-    for availability_domain in region.availability_domains:
+    for availability_domain in region.availability_domains_in(compartment_id):
         result[availability_domain.name] = list_shapes_in_domain(
             availability_domain.name, region.compute_client, compartment_id
         )
@@ -214,7 +214,7 @@ def check_availability_per_domain(
     all_shapes = set(shape_names)
     available_shapes_per_domain = {}
 
-    for availability_domain in region.availability_domains:
+    for availability_domain in region.availability_domains_in(compartment_id):
         shapes_to_check = {
             shape
             for shape in all_shapes
@@ -277,7 +277,7 @@ def get_available_domains(
     `shape_name` is available and within `shapes_quota`.
     """
     domains = []
-    for domain in region.availability_domains:
+    for domain in region.availability_domains_in(compartment_id):
         if shapes_quota.is_within_domain_quota(
             shape_name, domain.name
         ) and check_availability_in_domain(

src/tests/_internal/core/backends/oci/test_region.py

@@ -0,0 +1,54 @@
+from unittest.mock import MagicMock, patch
+
+from dstack._internal.core.backends.oci.region import OCIRegionClient
+
+CREDS_TENANCY = "ocid1.tenancy.oc1..credentials"
+OTHER_TENANCY_COMPARTMENT = "ocid1.compartment.oc1..shared-in-from-another-tenancy"
+
+
+def make_region_client() -> OCIRegionClient:
+    return OCIRegionClient(
+        {
+            "tenancy": CREDS_TENANCY,
+            "region": "us-phoenix-1",
+            "user": "ocid1.user.oc1..aaaa",
+            "fingerprint": "00:11:22",
+            "key_file": "/dev/null",
+        }
+    )
+
+
+class TestAvailabilityDomainsIn:
+    def test_resolves_domains_from_the_compartment_not_the_credentials(self):
+        region = make_region_client()
+        identity_client = MagicMock()
+        with patch.object(
+            OCIRegionClient, "identity_client", new_callable=lambda: identity_client
+        ):
+            region.availability_domains_in(OTHER_TENANCY_COMPARTMENT)
+
+        identity_client.list_availability_domains.assert_called_once_with(
+            OTHER_TENANCY_COMPARTMENT
+        )
+
+    def test_availability_domains_defaults_to_the_credentials_tenancy(self):
+        region = make_region_client()
+        identity_client = MagicMock()
+        with patch.object(
+            OCIRegionClient, "identity_client", new_callable=lambda: identity_client
+        ):
+            region.availability_domains
+
+        identity_client.list_availability_domains.assert_called_once_with(CREDS_TENANCY)
+
+    def test_result_is_cached_per_compartment(self):
+        region = make_region_client()
+        identity_client = MagicMock()
+        with patch.object(
+            OCIRegionClient, "identity_client", new_callable=lambda: identity_client
+        ):
+            region.availability_domains_in(OTHER_TENANCY_COMPARTMENT)
+            region.availability_domains_in(OTHER_TENANCY_COMPARTMENT)
+            region.availability_domains_in(CREDS_TENANCY)
+
+        assert identity_client.list_availability_domains.call_count == 2