runner failling should imply the runner is dead

this simplifies some restart logic that caused the runners to loop.
we only create runners if we failed or the instance has no failed
runners (the runners were revived successfully)

Author: Evanev7

src/exo/worker/plan.py

@@ -59,7 +59,7 @@ def plan(
     return (
         _cancel_tasks(runners, tasks)
         or _kill_runner(runners, all_runners, instances)
-        or _create_runner(node_id, runners, instances, instance_backoff)
+        or _create_runner(node_id, runners, all_runners, instances, instance_backoff)
         or _model_needs_download(
             node_id, runners, global_download_status, download_backoff
         )
@@ -96,22 +96,26 @@ def _kill_runner(
 def _create_runner(
     node_id: NodeId,
     runners: Mapping[RunnerId, RunnerSupervisor],
+    all_runners: Mapping[RunnerId, RunnerStatus],
     instances: Mapping[InstanceId, Instance],
     instance_backoff: KeyedBackoff[InstanceId],
 ) -> CreateRunner | None:
     for instance in instances.values():
-        if not instance_backoff.should_proceed(instance.instance_id):
-            continue
-
         runner_id = instance.shard_assignments.node_to_runner.get(node_id, None)
         if runner_id is None:
             continue
 
         if runner_id in runners:
             continue
 
-        shard = instance.shard(runner_id)
-        assert shard is not None
+        # don't create runners if any other nodes have runners that have failed - wait for them to fix themselves first.
+        instance_has_failed_runner = any( isinstance(all_runners.get(remote_runner_id), RunnerFailed) for remote_runner_id in instance.shard_assignments.node_to_runner.values() if remote_runner_id != runner_id )
+        we_have_failed_before = isinstance(all_runners.get(runner_id), RunnerFailed)
+        if instance_has_failed_runner and not we_have_failed_before:
+            continue
+
+        if not instance_backoff.should_proceed(instance.instance_id):
+            continue
 
         return CreateRunner(
             instance_id=instance.instance_id,

src/exo/worker/runner/image_models/runner.py

@@ -332,7 +332,7 @@ def main(self):
     def handle_task(self, task: Task):
         match task:
             case ConnectToGroup() if isinstance(
-                self.current_status, (RunnerIdle, RunnerFailed)
+                self.current_status, RunnerIdle
             ):
                 logger.info("runner connecting")
                 self.update_status(RunnerConnecting())

src/exo/worker/runner/llm_inference/runner.py

@@ -150,7 +150,7 @@ def handle_first_task(self, task: Task):
 
         match task:
             case ConnectToGroup() if isinstance(
-                self.current_status, (RunnerIdle, RunnerFailed)
+                self.current_status, RunnerIdle
             ):
                 assert isinstance(self.generator, Builder)
                 logger.info("runner connecting")