Forbid new fleets creation with fleets specified

Author: r4victor

src/dstack/_internal/server/background/tasks/process_submitted_jobs.py

@@ -290,6 +290,16 @@ async def _process_submitted_job(session: AsyncSession, job_model: JobModel):
                 master_job_provisioning_data=master_job_provisioning_data,
                 volumes=volumes,
             )
+            if fleet_model is None and run_spec.configuration.fleets is not None:
+                # Run cannot create new fleets when fleets are specified
+                logger.debug("%s: failed to use specified fleets", fmt(job_model))
+                job_model.status = JobStatus.TERMINATING
+                job_model.termination_reason = (
+                    JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY
+                )
+                job_model.last_processed_at = common_utils.get_current_datetime()
+                await session.commit()
+                return
             instance = await _assign_job_to_fleet_instance(
                 session=session,
                 instances_with_offers=fleet_instances_with_offers,

src/dstack/_internal/server/testing/common.py

@@ -258,8 +258,8 @@ async def create_file_archive(
 
 
 def get_run_spec(
-    run_name: str,
     repo_id: str,
+    run_name: str = "test-run",
     configuration_path: str = "dstack.yaml",
     profile: Union[Profile, Callable[[], Profile], None] = lambda: Profile(name="default"),
     configuration: Optional[AnyRunConfiguration] = None,

src/tests/_internal/server/background/tasks/test_process_submitted_jobs.py

@@ -732,12 +732,47 @@ async def test_assigns_no_fleet_when_all_fleets_occupied(self, test_db, session:
         await session.commit()
         await process_submitted_jobs()
         await session.refresh(job)
-        res = await session.execute(select(JobModel).options(joinedload(JobModel.instance)))
-        job = res.unique().scalar_one()
         assert job.status == JobStatus.SUBMITTED
         assert job.instance_assigned
-        assert job.instance is None
-        assert job.fleet is None
+        assert job.instance_id is None
+        assert job.fleet_id is None
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)
+    async def test_fails_with_no_capacity_when_specified_fleets_occupied(
+        self, test_db, session: AsyncSession
+    ):
+        project = await create_project(session)
+        user = await create_user(session)
+        repo = await create_repo(session=session, project_id=project.id)
+        fleet = await create_fleet(session=session, project=project, name="test-fleet")
+        instance = await create_instance(
+            session=session,
+            project=project,
+            fleet=fleet,
+            instance_num=0,
+            status=InstanceStatus.BUSY,
+        )
+        fleet.instances.append(instance)
+        run_spec = get_run_spec(repo_id=repo.name)
+        run_spec.configuration.fleets = ["test-fleet"]
+        run = await create_run(
+            session=session,
+            project=project,
+            repo=repo,
+            user=user,
+            run_spec=run_spec,
+        )
+        job = await create_job(
+            session=session,
+            run=run,
+            instance_assigned=False,
+        )
+        await session.commit()
+        await process_submitted_jobs()
+        await session.refresh(job)
+        assert job.status == JobStatus.TERMINATING
+        assert job.termination_reason == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)
@@ -770,8 +805,6 @@ async def test_assigns_job_to_elastic_empty_fleet(self, test_db, session: AsyncS
         )
         await process_submitted_jobs()
         await session.refresh(job)
-        res = await session.execute(select(JobModel))
-        job = res.unique().scalar_one()
         assert job.status == JobStatus.SUBMITTED
         assert job.instance_assigned
 
@@ -867,8 +900,6 @@ async def test_assigns_job_to_optimal_fleet(self, test_db, session: AsyncSession
         )
         await process_submitted_jobs()
         await session.refresh(job)
-        res = await session.execute(select(JobModel))
-        job = res.unique().scalar_one()
         assert job.status == JobStatus.SUBMITTED
         assert job.instance_assigned
         assert job.fleet_id == fleet2.id