fix(iris): add --async to gcloud delete/reset to unblock heartbeat thread

Synchronous gcloud subprocess calls (especially `tpu-vm delete`, 10-18s)
were blocking the heartbeat thread, stalling all task dispatch cluster-wide
for minutes at a time. Add `--async` flag to all gcloud delete and reset
commands so they return immediately after submitting the operation to GCP.

All callers already handle terminate() failures gracefully (catch + log),
so async deletion is safe. Create and describe commands remain synchronous
since their results are needed immediately.

Fixes #3678

Co-authored-by: Russell Power <rjpower@users.noreply.github.com>

Author: github-actions[bot]

lib/iris/src/iris/cluster/platform/gcp.py

@@ -373,6 +373,7 @@ def reboot(self) -> None:
             f"--project={self._project_id}",
             f"--zone={self._zone}",
             "--quiet",
+            "--async",
         ]
         logger.info("Rebooting GCE instance: %s", self._gce_vm_name)
         logger.info("gcloud command: %s", cmd)
@@ -388,8 +389,9 @@ def terminate(self) -> None:
             f"--project={self._project_id}",
             f"--zone={self._zone}",
             "--quiet",
+            "--async",
         ]
-        logger.info("Deleting GCE instance: %s", self._gce_vm_name)
+        logger.info("Deleting GCE instance (async): %s", self._gce_vm_name)
         logger.info("gcloud command: %s", cmd)
         result = subprocess.run(cmd, capture_output=True, text=True)
         if result.returncode != 0:
@@ -585,8 +587,9 @@ def terminate(self) -> None:
             f"--zone={self._zone}",
             f"--project={self._project_id}",
             "--quiet",
+            "--async",
         ]
-        logger.info("Terminating TPU: %s", self._slice_id)
+        logger.info("Terminating TPU (async): %s", self._slice_id)
         logger.info("gcloud command: %s", cmd)
         result = subprocess.run(cmd, capture_output=True, text=True)
         if result.returncode != 0:
@@ -722,8 +725,9 @@ def terminate(self) -> None:
             f"--project={self._project_id}",
             f"--zone={self._zone}",
             "--quiet",
+            "--async",
         ]
-        logger.info("Terminating VM slice: %s (vm=%s)", self._slice_id, self._vm_name)
+        logger.info("Terminating VM slice (async): %s (vm=%s)", self._slice_id, self._vm_name)
         logger.info("gcloud command: %s", cmd)
         result = subprocess.run(cmd, capture_output=True, text=True)
         if result.returncode != 0:
@@ -785,6 +789,7 @@ def _best_effort_delete_tpu(self, slice_id: str, zone: str) -> None:
         """Try to delete a TPU VM that may have been partially created.
 
         Silently ignores "not found" errors (resource was never created).
+        Uses --async so the caller is not blocked waiting for deletion.
         """
         cmd = [
             "gcloud",
@@ -796,8 +801,9 @@ def _best_effort_delete_tpu(self, slice_id: str, zone: str) -> None:
             f"--zone={zone}",
             f"--project={self._project_id}",
             "--quiet",
+            "--async",
         ]
-        logger.info("Best-effort cleanup of TPU %s in %s", slice_id, zone)
+        logger.info("Best-effort async cleanup of TPU %s in %s", slice_id, zone)
         result = subprocess.run(cmd, capture_output=True, text=True)
         if result.returncode != 0:
             error = result.stderr.strip()
@@ -808,6 +814,7 @@ def _best_effort_delete_vm(self, vm_name: str, zone: str) -> None:
         """Try to delete a GCE VM that may have been partially created.
 
         Silently ignores "not found" errors (resource was never created).
+        Uses --async so the caller is not blocked waiting for deletion.
         """
         cmd = [
             "gcloud",
@@ -818,8 +825,9 @@ def _best_effort_delete_vm(self, vm_name: str, zone: str) -> None:
             f"--zone={zone}",
             f"--project={self._project_id}",
             "--quiet",
+            "--async",
         ]
-        logger.info("Best-effort cleanup of VM %s in %s", vm_name, zone)
+        logger.info("Best-effort async cleanup of VM %s in %s", vm_name, zone)
         result = subprocess.run(cmd, capture_output=True, text=True)
         if result.returncode != 0:
             error = result.stderr.strip()