[iris] Use lightweight GetJobState for Fray actor polling

FrayActorJob.wait_ready and is_done were calling the heavy GetJobStatus RPC on
every 0.5s tick just to read .state, pairing 1:1 with the GetJobState polls from
wait_for_job. Switch both to the lightweight state RPC, back off from 0.1s to
5s between polls, and add IrisClient.job_state(job_id) plus wire Job.state
through it. Only fetches the full JobStatus on the terminal-error path where
the error message is actually needed.

Author: rjpower

lib/fray/src/fray/v2/iris_backend.py

@@ -36,6 +36,7 @@
 from iris.cluster.types import CoschedulingConfig, EnvironmentSpec, ResourceSpec, is_job_finished
 from iris.cluster.types import Entrypoint as IrisEntrypoint
 from iris.rpc import job_pb2
+from rigging.timing import ExponentialBackoff
 
 from fray.v2.actor import (
     ActorContext,
@@ -477,7 +478,7 @@ def wait_ready(self, count: int | None = None, timeout: float = 900.0) -> list[A
         """
         target = count if count is not None else self._count
         start = time.monotonic()
-        sleep_secs = 0.5
+        backoff = ExponentialBackoff(initial=0.1, maximum=5.0)
 
         while True:
             self.discover_new(target=target)
@@ -487,28 +488,29 @@ def wait_ready(self, count: int | None = None, timeout: float = 900.0) -> list[A
 
             # Fail fast if the underlying job has terminated (e.g. crash, OOM,
             # missing interpreter). Without this check we'd spin for the full
-            # timeout waiting for endpoints that will never appear.
+            # timeout waiting for endpoints that will never appear. Use the
+            # lightweight state-only RPC and fetch the full status only when we
+            # actually need the error message.
             client = self._get_client()
-            job_status = client.status(self._job_id)
-            if is_job_finished(job_status.state):
-                error = job_status.error or "unknown error"
+            state = client.job_state(self._job_id)
+            if is_job_finished(state):
+                error = client.status(self._job_id).error or "unknown error"
                 raise RuntimeError(
                     f"Actor job {self._job_id} finished before all actors registered "
                     f"({len(self._discovered_names)}/{target} ready). "
-                    f"Job state={job_status.state}, error={error}"
+                    f"Job state={state}, error={error}"
                 )
 
             elapsed = time.monotonic() - start
             if elapsed >= timeout:
                 raise TimeoutError(f"Only {len(self._discovered_names)}/{target} actors ready after {timeout}s")
 
-            time.sleep(sleep_secs)
+            time.sleep(backoff.next_interval())
 
     def is_done(self) -> bool:
         """Return True if the Iris worker job has permanently terminated."""
         client = self._get_client()
-        job_status = client.status(self._job_id)
-        return is_job_finished(job_status.state)
+        return is_job_finished(client.job_state(self._job_id))
 
     def shutdown(self) -> None:
         """Terminate the actor job."""

lib/iris/src/iris/client/client.py

@@ -23,7 +23,7 @@
 from contextvars import ContextVar
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Protocol
+from typing import Protocol, cast
 
 from connectrpc.code import Code
 from connectrpc.errors import ConnectError
@@ -234,8 +234,8 @@ def state_only(self) -> int:
 
     @property
     def state(self) -> job_pb2.JobState:
-        """Get current job state (shortcut for status().state)."""
-        return self.status().state
+        """Get current job state via the lightweight state-only RPC."""
+        return cast(job_pb2.JobState, self.state_only())
 
     def tasks(self) -> list[Task]:
         """Get all tasks for this job.
@@ -706,6 +706,17 @@ def status(self, job_id: JobName) -> job_pb2.JobStatus:
         """
         return self._cluster_client.get_job_status(job_id)
 
+    def job_state(self, job_id: JobName) -> int:
+        """Lightweight state query that avoids loading tasks/attempts/workers.
+
+        Prefer this over ``status(job_id).state`` for polling loops.
+        """
+        states = self._cluster_client.get_job_states([job_id])
+        wire_id = job_id.to_wire()
+        if wire_id not in states:
+            raise KeyError(f"Job {wire_id} not found")
+        return states[wire_id]
+
     def terminate(self, job_id: JobName) -> None:
         """Terminate a running job.