Fixes.

Author: rjpower

lib/iris/pyproject.toml

@@ -81,7 +81,8 @@ packages = ["src/iris"]
 
 [tool.pytest.ini_options]
 timeout = 10
-addopts = "-n auto --durations=25 -m 'not slow and not docker' -v"
+timeout_method = "thread"
+addopts = "-n auto --durations=25 -m 'not slow and not docker and not e2e' -v"
 markers = [
     "slow: marks tests as slow (deselect with '-m \"not slow\"')",
     "docker: marks tests requiring Docker runtime (slow, needs daemon)",

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

@@ -2236,9 +2236,13 @@ def _run_ping_loop(self, stop_event: threading.Event) -> None:
         worker. This is safe: fail_workers_batch, on_worker_failed, and
         terminate_slices_for_workers are all idempotent.
         """
-        limiter = RateLimiter(interval_seconds=self._config.heartbeat_interval.to_seconds())
+        ping_interval_s = self._config.heartbeat_interval.to_seconds()
+        limiter = RateLimiter(interval_seconds=ping_interval_s)
         ping_failures: dict[str, int] = {}
         threshold = self._config.heartbeat_failure_threshold
+        # Refresh resource snapshots every ~60s; other cycles just note liveness.
+        resource_update_every = max(1, round(60.0 / ping_interval_s))
+        cycle = 0
 
         while not stop_event.is_set():
             if not limiter.wait(cancel=stop_event):
@@ -2249,8 +2253,11 @@ def _run_ping_loop(self, stop_event: threading.Event) -> None:
                 self._reap_stale_workers()
                 workers = self._get_active_worker_addresses()
                 results = self._provider.ping_workers(workers)
+                update_resources = cycle % resource_update_every == 0
+                cycle += 1
 
                 dead_workers: list[str] = []
+                liveness_ids: list[WorkerId] = []
                 for result in results:
                     wid_str = str(result.worker_id)
                     if result.error is not None:
@@ -2264,8 +2271,13 @@ def _run_ping_loop(self, stop_event: threading.Event) -> None:
                             )
                     else:
                         ping_failures.pop(wid_str, None)
-                        if result.resource_snapshot:
+                        if update_resources and result.resource_snapshot:
                             self._transitions.update_worker_ping_success(result.worker_id, result.resource_snapshot)
+                        else:
+                            liveness_ids.append(result.worker_id)
+
+                if liveness_ids:
+                    self._transitions.touch_worker_liveness(liveness_ids)
 
                 if dead_workers:
                     failure_result = self._transitions.fail_workers_batch(
@@ -2302,10 +2314,10 @@ def _run_ping_loop(self, stop_event: threading.Event) -> None:
     def _run_poll_loop(self, stop_event: threading.Event) -> None:
         """Periodic full-state reconciliation for split heartbeat mode.
 
-        Polls all workers via PollTasks every 30s and feeds results into the
+        Polls all workers via PollTasks every 60s and feeds results into the
         task-updater queue for batched application.
         """
-        limiter = RateLimiter(interval_seconds=30.0)
+        limiter = RateLimiter(interval_seconds=60.0)
         while not stop_event.is_set():
             if not limiter.wait(cancel=stop_event):
                 break

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

@@ -13,7 +13,7 @@
 import json
 import logging
 from dataclasses import dataclass, field
-from collections.abc import Callable, Iterable
+from collections.abc import Callable, Iterable, Sequence
 from typing import Any, NamedTuple
 
 from iris.cluster.constraints import AttributeValue, Constraint, constraints_from_resources, merge_constraints
@@ -3042,6 +3042,17 @@ def _stopped() -> bool:
     # Split Heartbeat Helpers
     # =========================================================================
 
+    def touch_worker_liveness(self, worker_ids: Sequence[WorkerId]) -> None:
+        """Cheap liveness bump: update last_heartbeat_ms without rewriting resources."""
+        if not worker_ids:
+            return
+        now_ms = Timestamp.now().epoch_ms()
+        with self._db.transaction() as cur:
+            cur.executemany(
+                "UPDATE workers SET last_heartbeat_ms = ? WHERE worker_id = ?",
+                [(now_ms, str(wid)) for wid in worker_ids],
+            )
+
     def update_worker_ping_success(self, worker_id: WorkerId, resource_snapshot: job_pb2.WorkerResourceSnapshot) -> None:
         """Update worker timestamp and resource snapshot from a successful ping.