[UX] Show No quota and No balance when provisioning fails

src/dstack/_internal/cli/services/configurators/run.py

@@ -894,7 +894,12 @@ def print_finished_message(run: Run):
         if (
             termination_reason
             # A run that never started has no runner logs to read.
-            and termination_reason != JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY.value
+            and termination_reason
+            not in (
+                JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY.value,
+                JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA.value,
+                JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE.value,
+            )
         ):
             console.print(f"Check [code]dstack logs -d {run.name}[/code] for more details.")
 

src/dstack/_internal/cli/utils/fleet.py

@@ -11,7 +11,11 @@
 from dstack._internal.core.models.backends.base import BackendType
 from dstack._internal.core.models.fleets import Fleet, FleetNodesSpec, FleetStatus
 from dstack._internal.core.models.health import HealthStatus
-from dstack._internal.core.models.instances import Instance, InstanceStatus
+from dstack._internal.core.models.instances import (
+    Instance,
+    InstanceStatus,
+    InstanceTerminationReason,
+)
 from dstack._internal.core.models.resources import GPUSpec, ResourcesSpec
 from dstack._internal.utils.common import DateFormatter, pretty_date
 
@@ -166,6 +170,11 @@ def get_fleets_table(
 
             if instance.status == InstanceStatus.TERMINATED and instance.termination_reason:
                 instance_row["ERROR"] = instance.termination_reason
+                if (
+                    instance.termination_reason == InstanceTerminationReason.NO_OFFERS
+                    and instance.termination_reason_message in ("No quota", "No balance")
+                ):
+                    instance_row["ERROR"] = instance.termination_reason_message.lower()
 
             # No row-level dimming: only the columns styled above are dimmed on instance rows
             add_row_from_dict(table, instance_row)

src/dstack/_internal/core/models/instances.py

@@ -246,6 +246,8 @@ class InstanceTerminationReason(str, Enum):
     JOB_FINISHED = "job_finished"
     UNREACHABLE = "unreachable"
     NO_OFFERS = "no_offers"
+    # Add read support before emission so server replicas can be upgraded gradually.
+    NO_QUOTA = "no_quota"
     MASTER_FAILED = "master_failed"
     MAX_INSTANCES_LIMIT = "max_instances_limit"
     FLEET_SPEC_MISMATCH = "fleet_spec_mismatch"

src/dstack/_internal/core/models/runs.py

@@ -138,6 +138,9 @@ def to_error(self) -> Optional[str]:
 class JobTerminationReason(str, Enum):
     # Set by the server
     FAILED_TO_START_DUE_TO_NO_CAPACITY = "failed_to_start_due_to_no_capacity"
+    # Add read support before emission so server replicas can be upgraded gradually.
+    FAILED_TO_START_DUE_TO_NO_QUOTA = "failed_to_start_due_to_no_quota"
+    FAILED_TO_START_DUE_TO_NO_BALANCE = "failed_to_start_due_to_no_balance"
     INTERRUPTED_BY_NO_CAPACITY = "interrupted_by_no_capacity"
     INSTANCE_UNREACHABLE = "instance_unreachable"
     INSTANCE_ACCESS_REVOKED = "instance_access_revoked"
@@ -163,6 +166,8 @@ class JobTerminationReason(str, Enum):
     def to_status(self) -> JobStatus:
         mapping = {
             self.FAILED_TO_START_DUE_TO_NO_CAPACITY: JobStatus.FAILED,
+            self.FAILED_TO_START_DUE_TO_NO_QUOTA: JobStatus.FAILED,
+            self.FAILED_TO_START_DUE_TO_NO_BALANCE: JobStatus.FAILED,
             self.INTERRUPTED_BY_NO_CAPACITY: JobStatus.FAILED,
             self.INSTANCE_UNREACHABLE: JobStatus.FAILED,
             self.INSTANCE_ACCESS_REVOKED: JobStatus.FAILED,
@@ -194,6 +199,9 @@ def to_retry_event(self) -> Optional[RetryEvent]:
         """
         mapping = {
             self.FAILED_TO_START_DUE_TO_NO_CAPACITY: RetryEvent.NO_CAPACITY,
+            # Preserve existing no-capacity retry configurations for blocked offers.
+            self.FAILED_TO_START_DUE_TO_NO_QUOTA: RetryEvent.NO_CAPACITY,
+            self.FAILED_TO_START_DUE_TO_NO_BALANCE: RetryEvent.NO_CAPACITY,
             self.INTERRUPTED_BY_NO_CAPACITY: RetryEvent.INTERRUPTION,
         }
         default = RetryEvent.ERROR if self.to_status() == JobStatus.FAILED else None

src/dstack/_internal/server/background/pipeline_tasks/fleets.py

@@ -819,8 +819,8 @@ def _set_fail_instances_on_master_bootstrap_failure(
     instance_id_to_update_map: dict[uuid.UUID, _InstanceUpdateMap],
 ) -> None:
     """
-    Terminates instances with MASTER_FAILED if the master dies with NO_OFFERS in a cluster with node.min == 0.
-    This is needed to avoid master re-election loop and fail fast.
+    Terminates unprovisioned siblings when the cluster master has no usable offers.
+    For nodes.min == 0, this prevents a master re-election loop.
     """
     fleet_spec = get_fleet_spec(fleet_model)
     if (
@@ -841,7 +841,12 @@ def _set_fail_instances_on_master_bootstrap_failure(
 
     if (
         current_master_instance_model.status != InstanceStatus.TERMINATED
-        or current_master_instance_model.termination_reason != InstanceTerminationReason.NO_OFFERS
+        or current_master_instance_model.termination_reason
+        not in (
+            InstanceTerminationReason.NO_OFFERS,
+            InstanceTerminationReason.NO_QUOTA,
+            InstanceTerminationReason.NO_BALANCE,
+        )
     ):
         return
 

src/dstack/_internal/server/background/pipeline_tasks/instances/cloud_provisioning.py

@@ -23,6 +23,7 @@
     SkipOffer,
 )
 from dstack._internal.core.models.instances import (
+    InstanceAvailability,
     InstanceOfferWithAvailability,
     InstanceStatus,
     InstanceTerminationReason,
@@ -116,7 +117,24 @@ async def create_cloud_instance(instance_model: InstanceModel) -> ProcessResult:
         include_only_create_instance_supported_backends=True,
     )
 
-    offers_iter = iter(offers)
+    available_offers = offers.offers
+    for availability, message in (
+        (InstanceAvailability.NO_QUOTA, "No quota"),
+        (InstanceAvailability.NO_BALANCE, "No balance"),
+    ):
+        if offers.unavailable_reason == availability:
+            set_status_update(
+                update_map=result.instance_update_map,
+                instance_model=instance_model,
+                new_status=InstanceStatus.TERMINATED,
+                termination_reason=InstanceTerminationReason.NO_OFFERS,
+                # TODO: Store NO_QUOTA/NO_BALANCE once the minimum supported rolling-upgrade
+                # source version handles them on every replica.
+                termination_reason_message=message,
+            )
+            return result
+
+    offers_iter = iter(available_offers)
     offers_tried = 0
     # Limit number of offers tried to prevent long-running processing in case all offers fail.
     while offers_tried < server_settings.MAX_OFFERS_TRIED:
@@ -241,7 +259,7 @@ async def create_cloud_instance(instance_model: InstanceModel) -> ProcessResult:
         instance_model=instance_model,
         new_status=InstanceStatus.TERMINATED,
         termination_reason=InstanceTerminationReason.NO_OFFERS,
-        termination_reason_message="All offers failed" if offers else "No offers found",
+        termination_reason_message="All offers failed" if available_offers else "No offers found",
     )
     return result
 

src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py

@@ -31,6 +31,7 @@
     InstanceGroupPlacement,
 )
 from dstack._internal.core.models.instances import (
+    InstanceAvailability,
     InstanceOfferWithAvailability,
     InstanceStatus,
 )
@@ -485,7 +486,7 @@ class _NewCapacityAttempts:
 class _FailedNewCapacityProvisioning:
     placement_group_cleanup: Optional[_PlacementGroupCleanup]
     message: Optional[str] = None
-    """Why the job could not be provisioned. `None` if the offers were never tried."""
+    """Why the job could not be provisioned."""
 
 
 @dataclass
@@ -2420,8 +2421,21 @@ async def _provision_new_capacity(
         instance_mounts=check_run_spec_requires_instance_mounts(run.run_spec),
         placement_group=placement_group_model_to_placement_group_optional(placement_group_model),
     )
+    available_offers = offers.offers
+    for availability, message in (
+        (InstanceAvailability.NO_QUOTA, "No quota"),
+        (InstanceAvailability.NO_BALANCE, "No balance"),
+    ):
+        if offers.unavailable_reason == availability:
+            return _FailedNewCapacityProvisioning(
+                placement_group_cleanup=None,
+                # TODO: Store FAILED_TO_START_DUE_TO_NO_QUOTA/NO_BALANCE once the minimum
+                # supported rolling-upgrade source version handles them on every replica.
+                message=message,
+            )
+
     extra_authorized_keys = get_extra_authorized_keys(run.run_spec)
-    offers_iter = iter(offers)
+    offers_iter = iter(available_offers)
     offers_tried = 0
     offers_taken = 0
     skip_reasons: list[str] = []
@@ -2573,12 +2587,12 @@ async def _provision_new_capacity(
         message=_get_new_capacity_failure_message(
             fleet_name=fleet_model.name,
             attempts=_NewCapacityAttempts(
-                total=len(offers),
+                total=len(available_offers),
                 tried=offers_tried,
                 skip_reasons=skip_reasons,
                 errors=offer_errors,
                 # Offers are left only if the attempt limit, not the offer list, ended the loop.
-                limit_reached=offers_taken < len(offers),
+                limit_reached=offers_taken < len(available_offers),
             ),
         ),
         placement_group_cleanup=_build_placement_group_cleanup(

src/dstack/_internal/server/background/pipeline_tasks/runs/common.py

@@ -127,7 +127,7 @@ async def delete_superseded_no_capacity_job_submissions(
     new_job_models: list[JobModel],
 ) -> None:
     """
-    Delete previous job submissions that failed to start due to no capacity without
+    Delete previous job submissions that failed to start due to capacity, quota, or balance without
     ever provisioning. Such submissions are created each time a run is retrying
     on no capacity and carry no information beyond the events.
     The direct predecessor of each new submission is kept so that the run's
@@ -143,8 +143,13 @@ async def delete_superseded_no_capacity_job_submissions(
                 JobModel.job_num == job_model.job_num,
                 JobModel.submission_num < job_model.submission_num - 1,
                 JobModel.status.in_(JobStatus.finished_statuses()),
-                JobModel.termination_reason
-                == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY,
+                JobModel.termination_reason.in_(
+                    [
+                        JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY,
+                        JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA,
+                        JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE,
+                    ]
+                ),
                 JobModel.job_provisioning_data.is_(None),
             )
         )

src/dstack/_internal/server/services/backends/__init__.py

@@ -40,7 +40,6 @@
 from dstack._internal.core.models.runs import Requirements
 from dstack._internal.server import settings
 from dstack._internal.server.models import BackendModel, DecryptedString, ProjectModel
-from dstack._internal.server.services.offers import merge_offer_iterables
 from dstack._internal.utils.common import run_async
 from dstack._internal.utils.logging import get_logger
 
@@ -472,26 +471,17 @@ async def get_backend_offers(
     requirements: Requirements,
     full_offers: bool,
     unallocated_resources: bool,
-    exclude_not_available: bool = False,
-) -> Iterable[Tuple[Backend, InstanceOfferWithAvailability]]:
+) -> list[tuple[Backend, Iterable[InstanceOfferWithAvailability]]]:
     """
-    Returns a lazy iterable of backend offers satisfying `requirements` sorted by price.
+    Return each backend's lazy offers satisfying `requirements`.
     """
 
-    def get_filtered_offers_with_backends(
-        backend: Backend,
-        offers: Iterable[InstanceOfferWithAvailability],
-    ) -> Iterator[Tuple[Backend, InstanceOfferWithAvailability]]:
-        for offer in offers:
-            if not exclude_not_available or offer.availability.is_available():
-                yield (backend, offer)
-
     logger.debug("Requesting instance offers from backends: %s", [b.TYPE.value for b in backends])
     tasks = [
         run_async(get_offers_tracked, backend, requirements, full_offers, unallocated_resources)
         for backend in backends
     ]
-    offers_by_backend: list[Iterable[tuple[Backend, InstanceOfferWithAvailability]]] = []
+    offers_by_backend: list[tuple[Backend, Iterable[InstanceOfferWithAvailability]]] = []
     for backend, result in zip(backends, await asyncio.gather(*tasks, return_exceptions=True)):
         if isinstance(result, BackendError):
             logger.warning(
@@ -507,8 +497,8 @@ def get_filtered_offers_with_backends(
                 exc_info=result,
             )
             continue
-        offers_by_backend.append(get_filtered_offers_with_backends(backend, result))
-    return merge_offer_iterables(*offers_by_backend)
+        offers_by_backend.append((backend, result))
+    return offers_by_backend
 
 
 def check_backend_type_available(backend_type: BackendType):

src/dstack/_internal/server/services/fleets.py

@@ -3,13 +3,12 @@
 from collections.abc import Callable
 from datetime import datetime
 from functools import wraps
-from typing import List, Literal, Optional, Tuple, TypeVar, Union
+from typing import List, Literal, Optional, TypeVar, Union
 
 from sqlalchemy import and_, exists, false, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import aliased, joinedload, selectinload
 
-from dstack._internal.core.backends.base.backend import Backend
 from dstack._internal.core.backends.features import BACKENDS_WITH_CREATE_INSTANCE_SUPPORT
 from dstack._internal.core.errors import (
     ForbiddenError,
@@ -32,7 +31,6 @@
     SSHParams,
 )
 from dstack._internal.core.models.instances import (
-    InstanceOfferWithAvailability,
     InstanceStatus,
     InstanceTerminationReason,
     SSHConnectionParams,
@@ -470,7 +468,7 @@ async def get_plan(
                 include_only_create_instance_supported_backends
             ),
         )
-        offers = [offer for _, offer in offers_with_backends]
+        offers = [offer for _, offer in offers_with_backends.offers]
 
     _remove_fleet_spec_sensitive_info(effective_spec)
     if current_fleet is not None:
@@ -501,7 +499,7 @@ async def get_fleet_offers(
     master_job_provisioning_data: Optional[JobProvisioningData] = None,
     infer_master_job_provisioning_data_from_fleet_instances: bool = True,
     include_only_create_instance_supported_backends: bool = True,
-) -> List[Tuple[Backend, InstanceOfferWithAvailability]]:
+) -> offers_services.Offers:
     """
     Return offers for fleet planning and provisioning.
 
@@ -536,13 +534,10 @@ async def get_fleet_offers(
         master_job_provisioning_data=master_job_provisioning_data,
         placement_group=placement_group,
         blocks=blocks,
+        supported_backend_types=BACKENDS_WITH_CREATE_INSTANCE_SUPPORT
+        if include_only_create_instance_supported_backends
+        else None,
     )
-    if include_only_create_instance_supported_backends:
-        offers = [
-            (backend, offer)
-            for backend, offer in offers
-            if offer.backend in BACKENDS_WITH_CREATE_INSTANCE_SUPPORT
-        ]
     return offers
 
 

src/dstack/_internal/server/services/jobs/__init__.py

@@ -625,7 +625,15 @@ def _get_job_status_message(job_model: JobModel) -> str:
                 and "No matching fleet found" in job_model.termination_reason_message
             ):
                 return "no fleets"
+            if job_model.termination_reason_message in ("No quota", "No balance"):
+                return job_model.termination_reason_message.lower()
             return "no capacity"
+        elif job_model.termination_reason == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA:
+            return "no quota"
+        elif (
+            job_model.termination_reason == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE
+        ):
+            return "no balance"
         elif job_model.termination_reason == JobTerminationReason.INTERRUPTED_BY_NO_CAPACITY:
             return "interrupted"
         else: