Optimize process_fleets queries

Author: r4victor

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

@@ -7,7 +7,7 @@
 
 from dstack._internal.core.models.fleets import FleetStatus
 from dstack._internal.server.db import get_db, get_session_ctx
-from dstack._internal.server.models import FleetModel
+from dstack._internal.server.models import FleetModel, InstanceModel, JobModel, RunModel
 from dstack._internal.server.services.fleets import (
     is_fleet_empty,
     is_fleet_in_use,
@@ -64,9 +64,11 @@ async def _process_fleet(session: AsyncSession, fleet_model: FleetModel):
     res = await session.execute(
         select(FleetModel)
         .where(FleetModel.id == fleet_model.id)
-        .options(joinedload(FleetModel.project))
-        .options(joinedload(FleetModel.instances))
-        .options(joinedload(FleetModel.runs))
+        .options(joinedload(FleetModel.instances).load_only(InstanceModel.deleted))
+        .options(
+            joinedload(FleetModel.instances).joinedload(InstanceModel.jobs).load_only(JobModel.id)
+        )
+        .options(joinedload(FleetModel.runs).load_only(RunModel.status))
         .execution_options(populate_existing=True)
     )
     fleet_model = res.unique().scalar_one()

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

@@ -10,7 +10,7 @@
 from dstack._internal.core.models.profiles import parse_duration
 from dstack._internal.core.models.volumes import VolumeStatus
 from dstack._internal.server.db import get_db, get_session_ctx
-from dstack._internal.server.models import ProjectModel, VolumeModel
+from dstack._internal.server.models import ProjectModel, UserModel, VolumeModel
 from dstack._internal.server.services import backends as backends_services
 from dstack._internal.server.services.locking import get_locker
 from dstack._internal.server.services.volumes import (
@@ -49,7 +49,7 @@ async def process_idle_volumes():
             select(VolumeModel)
             .where(VolumeModel.id.in_(volume_ids))
             .options(joinedload(VolumeModel.project).joinedload(ProjectModel.backends))
-            .options(joinedload(VolumeModel.user))
+            .options(joinedload(VolumeModel.user).load_only(UserModel.name))
             .options(joinedload(VolumeModel.attachments))
             .execution_options(populate_existing=True)
         )

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

@@ -2,12 +2,14 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from dstack._internal.core.models.fleets import FleetStatus
+from dstack._internal.core.models.instances import InstanceStatus
 from dstack._internal.core.models.runs import RunStatus
 from dstack._internal.core.models.users import GlobalRole, ProjectRole
 from dstack._internal.server.background.tasks.process_fleets import process_fleets
 from dstack._internal.server.services.projects import add_project_member
 from dstack._internal.server.testing.common import (
     create_fleet,
+    create_instance,
     create_project,
     create_repo,
     create_run,
@@ -75,3 +77,27 @@ async def test_does_not_delete_fleet_with_active_run(self, test_db, session: Asy
         await process_fleets()
         await session.refresh(fleet)
         assert not fleet.deleted
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)
+    async def test_does_not_delete_fleet_with_instance(self, test_db, session: AsyncSession):
+        project = await create_project(session)
+        fleet = await create_fleet(
+            session=session,
+            project=project,
+        )
+        user = await create_user(session=session, global_role=GlobalRole.USER)
+        await add_project_member(
+            session=session, project=project, user=user, project_role=ProjectRole.USER
+        )
+        instance = await create_instance(
+            session=session,
+            project=project,
+            fleet=fleet,
+            status=InstanceStatus.IDLE,
+        )
+        fleet.instances.append(instance)
+        await session.commit()
+        await process_fleets()
+        await session.refresh(fleet)
+        assert not fleet.deleted