alternative: resume if RetireWorkers stopped

distributed/active_memory_manager.py

@@ -661,8 +661,6 @@ def run(self) -> SuggestionGenerator:
 
     def done(self) -> bool:
         """Return True if it is safe to close the worker down; False otherwise"""
-        if self not in self.manager.policies:
-            return True
         ws = self.manager.scheduler.workers.get(self.address)
         if ws is None:
             return True

distributed/scheduler.py

@@ -6071,8 +6071,15 @@ async def _track_retire_worker(
         close_workers: bool,
         remove: bool,
     ) -> tuple:  # tuple[str | None, dict]
+        amm = self.extensions["amm"]
         while not policy.done():
-            if policy.no_recipients:
+            if stopped := policy not in amm.policies:
+                logger.warning(
+                    f"RetireWorkers for {policy.address} stopped, but not done. "
+                    f"Setting worker back to {prev_status.name!r}. Keys on worker: "
+                    f"{ {ts: ts.who_has for ts in ws.has_what} }"
+                )
+            if policy.no_recipients or stopped:
                 # Abort retirement. This time we don't need to worry about race
                 # conditions and we can wait for a scheduler->worker->scheduler
                 # round-trip.