Prioritize pending deployment removal on scaling down

dask_kubernetes/operator/controller/controller.py

@@ -6,6 +6,7 @@
 from uuid import uuid4
 
 import aiohttp
+import anyio
 import dask.config
 import kopf
 import kr8s
@@ -606,27 +607,20 @@ async def daskworkergroup_replica_update(
             )
         logger.info(f"Scaled worker group {name} up to {desired_workers} workers.")
         if workers_needed < 0:
-            len_current_workers = len(current_workers)
-            deleted_workers = []
-            for idx in range(-workers_needed):
-                if idx > len_current_workers:
-                    break
-                worker = current_workers[idx]
-
-                is_running = False
-                for pod in await worker.pods():
-                    if pod.status.phase == "Running":
-                        is_running = True
-                        break
-
-                if not is_running:
-                    deleted_workers.append(worker)
-                    await worker.delete()
-
-            if deleted_workers:
-                logger.info(f"Deleted pending {len(deleted_workers)} workers")
-
-            n_workers = -workers_needed - len(deleted_workers)
+            unready_deployments = [
+                current_workers[idx]
+                for idx in range(-workers_needed)
+                if idx < len(current_workers) and not current_workers[idx].ready()
+            ]
+
+            async with anyio.create_task_group() as tg:
+                for deployment in unready_deployments:
+                    tg.start_soon(deployment.delete)
+
+            if unready_deployments:
+                logger.info(f"Deleted unready {len(unready_deployments)} workers.")
+
+            n_workers = -workers_needed - len(unready_deployments)
             if n_workers > 0:
                 worker_ids = await retire_workers(
                     n_workers=n_workers,