[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/core/models/instances.py

@@ -246,6 +246,7 @@ class InstanceTerminationReason(str, Enum):
     JOB_FINISHED = "job_finished"
     UNREACHABLE = "unreachable"
     NO_OFFERS = "no_offers"
+    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,8 @@ 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"
+    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 +165,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 +198,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,
@@ -45,7 +46,10 @@
     get_instance_requirements,
 )
 from dstack._internal.server.services.logging import fmt
-from dstack._internal.server.services.offers import get_instance_offer_with_restricted_az
+from dstack._internal.server.services.offers import (
+    UnavailableOffers,
+    get_instance_offer_with_restricted_az,
+)
 from dstack._internal.server.services.placement import (
     can_use_placement_groups,
     get_fleet_placement_group_models,
@@ -103,7 +107,8 @@ async def create_cloud_instance(instance_model: InstanceModel) -> ProcessResult:
         )
         master_job_provisioning_data = cluster_context.master_job_provisioning_data
 
-    offers = await get_fleet_offers(
+    unavailable_offers = UnavailableOffers()
+    available_offers = await get_fleet_offers(
         project=instance_model.project,
         profile=profile,
         requirements=requirements,
@@ -114,9 +119,33 @@ async def create_cloud_instance(instance_model: InstanceModel) -> ProcessResult:
         master_job_provisioning_data=master_job_provisioning_data,
         infer_master_job_provisioning_data_from_fleet_instances=False,
         include_only_create_instance_supported_backends=True,
+        unavailable_offers=unavailable_offers,
     )
 
-    offers_iter = iter(offers)
+    if not available_offers:
+        for availability, reason, message in (
+            (
+                InstanceAvailability.NO_QUOTA,
+                InstanceTerminationReason.NO_QUOTA,
+                "All matching offers are blocked by quota",
+            ),
+            (
+                InstanceAvailability.NO_BALANCE,
+                InstanceTerminationReason.NO_BALANCE,
+                "All matching offers are blocked by insufficient balance",
+            ),
+        ):
+            if unavailable_offers.all_have(availability):
+                set_status_update(
+                    update_map=result.instance_update_map,
+                    instance_model=instance_model,
+                    new_status=InstanceStatus.TERMINATED,
+                    termination_reason=reason,
+                    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 +270,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,
 )
@@ -117,6 +118,7 @@
 from dstack._internal.server.services.locking import get_locker
 from dstack._internal.server.services.logging import fmt
 from dstack._internal.server.services.offers import (
+    UnavailableOffers,
     get_instance_offer_with_restricted_az,
     get_offers_by_requirements,
 )
@@ -484,8 +486,9 @@ class _NewCapacityAttempts:
 @dataclass
 class _FailedNewCapacityProvisioning:
     placement_group_cleanup: Optional[_PlacementGroupCleanup]
+    reason: JobTerminationReason = JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY
     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
@@ -1494,7 +1497,7 @@ async def _process_new_capacity_provisioning(
     if isinstance(provision_new_capacity_result, _FailedNewCapacityProvisioning):
         logger.debug("%s: provisioning failed", fmt(context.job_model))
         return _TerminateSubmittedJobResult(
-            reason=JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY,
+            reason=provision_new_capacity_result.reason,
             message=provision_new_capacity_result.message,
             locked_fleet_id=locked_fleet_id,
             placement_group_cleanup=provision_new_capacity_result.placement_group_cleanup,
@@ -2408,7 +2411,8 @@ async def _provision_new_capacity(
         fleet_model=fleet_model,
     )
     multinode = requirements.multinode or is_multinode_job(job)
-    offers = await get_offers_by_requirements(
+    unavailable_offers = UnavailableOffers()
+    available_offers = await get_offers_by_requirements(
         project=project,
         profile=profile,
         requirements=requirements,
@@ -2419,9 +2423,30 @@ async def _provision_new_capacity(
         privileged=job.job_spec.privileged,
         instance_mounts=check_run_spec_requires_instance_mounts(run.run_spec),
         placement_group=placement_group_model_to_placement_group_optional(placement_group_model),
-    )
+        unavailable_offers=unavailable_offers,
+    )
+    if not available_offers:
+        for availability, reason, message in (
+            (
+                InstanceAvailability.NO_QUOTA,
+                JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA,
+                "All matching offers are blocked by quota",
+            ),
+            (
+                InstanceAvailability.NO_BALANCE,
+                JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE,
+                "All matching offers are blocked by insufficient balance",
+            ),
+        ):
+            if unavailable_offers.all_have(availability):
+                return _FailedNewCapacityProvisioning(
+                    placement_group_cleanup=None,
+                    reason=reason,
+                    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 +2598,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

@@ -473,6 +473,9 @@ async def get_backend_offers(
     full_offers: bool,
     unallocated_resources: bool,
     exclude_not_available: bool = False,
+    on_unavailable_offer: Optional[
+        Callable[[Backend, InstanceOfferWithAvailability], None]
+    ] = None,
 ) -> Iterable[Tuple[Backend, InstanceOfferWithAvailability]]:
     """
     Yields backend offers satisfying `requirements` sorted by price.
@@ -485,6 +488,8 @@ def get_filtered_offers_with_backends(
         for offer in offers:
             if not exclude_not_available or offer.availability.is_available():
                 yield (backend, offer)
+            elif on_unavailable_offer is not None:
+                on_unavailable_offer(backend, offer)
 
     logger.debug("Requesting instance offers from backends: %s", [b.TYPE.value for b in backends])
     tasks = [

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

@@ -501,6 +501,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,
+    unavailable_offers: Optional[offers_services.UnavailableOffers] = None,
 ) -> List[Tuple[Backend, InstanceOfferWithAvailability]]:
     """
     Return offers for fleet planning and provisioning.
@@ -536,13 +537,11 @@ async def get_fleet_offers(
         master_job_provisioning_data=master_job_provisioning_data,
         placement_group=placement_group,
         blocks=blocks,
+        unavailable_offers=unavailable_offers,
+        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

@@ -626,6 +626,12 @@ def _get_job_status_message(job_model: JobModel) -> str:
             ):
                 return "no fleets"
             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: