iris: remove unused _heartbeat_lock (#4835)

* drop `_heartbeat_lock` from `Controller` and the two `with`-blocks
that use it (around `_sync_all_execution_units` and
`_sync_direct_provider`)
* only one thread ever acquired it: `start()` spawns either
`_run_provider_loop` or `_run_direct_provider_loop`, never both, and
each loop round runs sequentially
* checkpointing no longer contends for the lock either[^1], so every
acquisition was uncontended

[^1]: backup reads via a dedicated RO connection and each heartbeat
round commits as an atomic batch

Co-authored-by: Rafal Wojdyla <ravwojdyla@gmail.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: 3 people

lib/iris/src/iris/cluster/controller/controller.py

@@ -1126,12 +1126,6 @@ def __init__(
 
         self._atexit_registered = False
 
-        # Serializes provider sync rounds so only one heartbeat round mutates
-        # DB state at a time. Checkpointing no longer takes this lock: the
-        # backup reads via a dedicated RO connection and each heartbeat round
-        # commits as an atomic batch, so snapshots are always consistent.
-        self._heartbeat_lock = threading.Lock()
-
         # Rate-limits periodic (best-effort) checkpoint writes.
         # None when checkpoint_interval is not configured.
         self._periodic_checkpoint_limiter: RateLimiter | None = (
@@ -1408,8 +1402,7 @@ def _run_provider_loop(self, stop_event: threading.Event) -> None:
             if stop_event.is_set():
                 break
             try:
-                with self._heartbeat_lock:
-                    self._sync_all_execution_units()
+                self._sync_all_execution_units()
             except Exception:
                 logger.exception("Provider sync round failed, will retry next interval")
 
@@ -1432,19 +1425,18 @@ def _sync_direct_provider(self) -> None:
             return
         assert isinstance(self._provider, K8sTaskProvider)
         provider = self._provider
-        with self._heartbeat_lock:
-            max_promotions = self._promotion_bucket.available
-            batch = self._transitions.drain_for_direct_provider(
-                max_promotions=max_promotions,
-            )
-            if batch.tasks_to_run:
-                self._promotion_bucket.try_acquire(len(batch.tasks_to_run))
-            result = provider.sync(batch)
-            tx_result = self._transitions.apply_direct_provider_updates(result.updates)
-            self._provider_scheduling_events = list(result.scheduling_events) if result.scheduling_events else []
-            self._provider_capacity = result.capacity
-            if tx_result.tasks_to_kill:
-                self.kill_tasks_on_workers(tx_result.tasks_to_kill, tx_result.task_kill_workers)
+        max_promotions = self._promotion_bucket.available
+        batch = self._transitions.drain_for_direct_provider(
+            max_promotions=max_promotions,
+        )
+        if batch.tasks_to_run:
+            self._promotion_bucket.try_acquire(len(batch.tasks_to_run))
+        result = provider.sync(batch)
+        tx_result = self._transitions.apply_direct_provider_updates(result.updates)
+        self._provider_scheduling_events = list(result.scheduling_events) if result.scheduling_events else []
+        self._provider_capacity = result.capacity
+        if tx_result.tasks_to_kill:
+            self.kill_tasks_on_workers(tx_result.tasks_to_kill, tx_result.task_kill_workers)
 
     def _run_profile_loop(self, stop_event: threading.Event) -> None:
         """Periodically capture CPU and memory profiles for all running tasks.