Use fleets with no instances if fleets specified

Author: r4victor

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

@@ -460,13 +460,11 @@ async def _refetch_fleet_models(
             *fleet_filters,
         )
         .where(
-            and_(
-                InstanceModel.id.in_(instances_ids),
-                or_(
-                    InstanceModel.id.is_(None),
-                    and_(
-                        *instance_filters,
-                    ),
+            or_(
+                InstanceModel.id.is_(None),
+                and_(
+                    InstanceModel.id.in_(instances_ids),
+                    *instance_filters,
                 ),
             )
         )
@@ -535,8 +533,12 @@ def _find_optimal_fleet_with_offers(
                 fleet_priority,
             )
         )
-    if all(t[2] == 0 for t in candidate_fleets_with_offers):
-        # If no fleets have available offers, create a new fleet.
+    if run_spec.configuration.fleets is None and all(
+        t[2] == 0 for t in candidate_fleets_with_offers
+    ):
+        # If fleets are not specified and no fleets have available offers, create a new fleet.
+        # This is for compatibility with non-fleet-first UX when runs created new fleets
+        # if there are no instances to reuse.
         return None, []
     candidate_fleets_with_offers.sort(key=lambda t: t[-1])
     return candidate_fleets_with_offers[0][:2]

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

@@ -776,15 +776,19 @@ async def test_fails_with_no_capacity_when_specified_fleets_occupied(
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)
-    async def test_assigns_job_to_elastic_empty_fleet(self, test_db, session: AsyncSession):
+    async def test_does_not_assign_job_to_elastic_empty_fleet_if_fleets_unspecified(
+        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_spec = get_fleet_spec()
         fleet_spec.configuration.nodes = Range(min=0, max=1)
-        await create_fleet(session=session, project=project, spec=fleet_spec)
+        await create_fleet(session=session, project=project, spec=fleet_spec, name="fleet")
         # Need a second non-empty fleet to have two-stage processing
-        fleet2 = await create_fleet(session=session, project=project, spec=fleet_spec)
+        fleet2 = await create_fleet(
+            session=session, project=project, spec=fleet_spec, name="fleet2"
+        )
         await create_instance(
             session=session,
             project=project,
@@ -807,6 +811,51 @@ async def test_assigns_job_to_elastic_empty_fleet(self, test_db, session: AsyncS
         await session.refresh(job)
         assert job.status == JobStatus.SUBMITTED
         assert job.instance_assigned
+        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_assigns_job_to_elastic_empty_fleet_if_fleets_specified(
+        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_spec = get_fleet_spec()
+        fleet_spec.configuration.nodes = Range(min=0, max=1)
+        fleet = await create_fleet(session=session, project=project, spec=fleet_spec, name="fleet")
+        # Need a second non-empty fleet to have two-stage processing
+        fleet2 = await create_fleet(
+            session=session, project=project, spec=fleet_spec, name="fleet2"
+        )
+        await create_instance(
+            session=session,
+            project=project,
+            fleet=fleet2,
+            instance_num=0,
+            status=InstanceStatus.BUSY,
+        )
+        run_spec = get_run_spec(repo_id=repo.name)
+        run_spec.configuration.fleets = [fleet.name, fleet2.name]
+        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 process_submitted_jobs()
+        await session.refresh(job)
+        assert job.status == JobStatus.SUBMITTED
+        assert job.instance_assigned
+        assert job.instance_id is None
+        assert job.fleet_id == fleet.id
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)
@@ -852,7 +901,6 @@ async def test_creates_new_instance_in_existing_empty_fleet(
         assert job.status == JobStatus.PROVISIONING
         assert job.instance is not None
         assert job.instance.instance_num == 0
-        assert job.instance.fleet_id == fleet.id
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)