Handle terminating deleted instances

Author: r4victor

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

@@ -11,7 +11,7 @@
 from pydantic import ValidationError
 from sqlalchemy import and_, delete, func, not_, select
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.orm import joinedload, with_loader_criteria
+from sqlalchemy.orm import joinedload
 
 from dstack._internal import settings
 from dstack._internal.core.backends.base.compute import (
@@ -218,9 +218,8 @@ async def _process_instance(session: AsyncSession, instance: InstanceModel):
             .options(joinedload(InstanceModel.project).joinedload(ProjectModel.backends))
             .options(joinedload(InstanceModel.jobs).load_only(JobModel.id, JobModel.status))
             .options(
-                joinedload(InstanceModel.fleet).joinedload(FleetModel.instances),
-                with_loader_criteria(
-                    InstanceModel, InstanceModel.deleted == False, include_aliases=True
+                joinedload(InstanceModel.fleet).joinedload(
+                    FleetModel.instances.and_(InstanceModel.deleted == False)
                 ),
             )
             .execution_options(populate_existing=True)
@@ -233,9 +232,8 @@ async def _process_instance(session: AsyncSession, instance: InstanceModel):
             .options(joinedload(InstanceModel.project))
             .options(joinedload(InstanceModel.jobs).load_only(JobModel.id, JobModel.status))
             .options(
-                joinedload(InstanceModel.fleet).joinedload(FleetModel.instances),
-                with_loader_criteria(
-                    InstanceModel, InstanceModel.deleted == False, include_aliases=True
+                joinedload(InstanceModel.fleet).joinedload(
+                    FleetModel.instances.and_(InstanceModel.deleted == False)
                 ),
             )
             .execution_options(populate_existing=True)

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

@@ -597,6 +597,34 @@ async def test_terminate(self, test_db, session: AsyncSession):
         assert instance.deleted_at is not None
         assert instance.finished_at is not None
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)
+    async def test_terminates_terminating_deleted_instance(self, test_db, session: AsyncSession):
+        # There was a race condition when instance could stay in Terminating while marked as deleted.
+        # TODO:
+        project = await create_project(session=session)
+        instance = await create_instance(
+            session=session, project=project, status=InstanceStatus.TERMINATING
+        )
+        instance.deleted = True
+        instance.termination_reason = InstanceTerminationReason.IDLE_TIMEOUT
+        instance.last_job_processed_at = instance.deleted_at = (
+            get_current_datetime() + dt.timedelta(minutes=-19)
+        )
+        await session.commit()
+
+        with self.mock_terminate_in_backend() as mock:
+            await process_instances()
+            mock.assert_called_once()
+
+        await session.refresh(instance)
+
+        assert instance is not None
+        assert instance.status == InstanceStatus.TERMINATED
+        assert instance.deleted == True
+        assert instance.deleted_at is not None
+        assert instance.finished_at is not None
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)
     @pytest.mark.parametrize(