Prevent failed instance retries (#1763)

## Motivation

Currently, when a runner fails, the master retries the instance. Most of
the time, this causes a loop over failure. Retries need backoff and a
cap.

## Changes

- src/exo/worker/main.py: Before creating a runner, check an exponential
backoff timer per instance. After EXO_MAX_INSTANCE_RETRIES failures,
send DeleteInstance to permanently remove the instance. Record attempts
on Shutdown; reset on InstanceDeleted.
- src/exo/utils/keyed_backoff.py: Add attempts() method to query retry
count
- src/exo/shared/constants.py: Add EXO_MAX_INSTANCE_RETRIES = 3.

## Why It Works

The worker gates CreateRunner tasks behind a KeyedBackoff, adding
exponential delay (2s base, 30s cap) between retries. After 3 failures
the worker sends DeleteInstance, stopping retries entirely. The backoff
resets when the instance is deleted, so a fresh placement starts clean.

---------

Co-authored-by: Evan <evanev7@gmail.com>

dashboard/src/lib/stores/app.svelte.ts

@@ -3256,31 +3256,6 @@ class AppStore {
     }
   }
 
-  /**
-   * Cancel/pause an active download on a specific node
-   */
-  async cancelDownload(nodeId: string, modelId: string): Promise<void> {
-    try {
-      const response = await fetch("/download/cancel", {
-        method: "POST",
-        headers: { "Content-Type": "application/json" },
-        body: JSON.stringify({
-          targetNodeId: nodeId,
-          modelId: modelId,
-        }),
-      });
-      if (!response.ok) {
-        const errorText = await response.text();
-        throw new Error(
-          `Failed to cancel download: ${response.status} - ${errorText}`,
-        );
-      }
-    } catch (error) {
-      console.error("Error cancelling download:", error);
-      throw error;
-    }
-  }
-
   /**
    * Delete a downloaded model from a specific node
    */
@@ -3502,8 +3477,6 @@ export const resetImageGenerationParams = () =>
 // Download actions
 export const startDownload = (nodeId: string, shardMetadata: object) =>
   appStore.startDownload(nodeId, shardMetadata);
-export const cancelDownload = (nodeId: string, modelId: string) =>
-  appStore.cancelDownload(nodeId, modelId);
 export const deleteDownload = (nodeId: string, modelId: string) =>
   appStore.deleteDownload(nodeId, modelId);
 

dashboard/src/routes/downloads/+page.svelte

@@ -9,7 +9,6 @@
     refreshState,
     lastUpdate as lastUpdateStore,
     startDownload,
-    cancelDownload,
     deleteDownload,
   } from "$lib/stores/app.svelte";
   import {
@@ -350,59 +349,6 @@
   });
 </script>
 
-{#snippet trashIcon()}
-  <svg
-    class="w-5 h-5"
-    viewBox="0 0 20 20"
-    fill="none"
-    stroke="currentColor"
-    stroke-width="2"
-  >
-    <path
-      d="M4 6h12M8 6V4h4v2m1 0v10a1 1 0 01-1 1H8a1 1 0 01-1-1V6h6"
-      stroke-linecap="round"
-      stroke-linejoin="round"
-    ></path>
-  </svg>
-{/snippet}
-
-{#snippet downloadIcon(size?: string)}
-  <svg
-    class={size ?? "w-5 h-5"}
-    viewBox="0 0 20 20"
-    fill="none"
-    stroke="currentColor"
-    stroke-width="2"
-  >
-    <path
-      d="M10 3v10m0 0l-3-3m3 3l3-3M3 17h14"
-      stroke-linecap="round"
-      stroke-linejoin="round"
-    ></path>
-  </svg>
-{/snippet}
-
-{#snippet pauseIcon()}
-  <svg class="w-5 h-5" viewBox="0 0 20 20" fill="currentColor">
-    <path
-      fill-rule="evenodd"
-      d="M6 4h2v12H6V4zm6 0h2v12h-2V4z"
-      clip-rule="evenodd"
-    ></path>
-  </svg>
-{/snippet}
-
-{#snippet deleteButton(nodeId: string, modelId: string)}
-  <button
-    type="button"
-    class="text-white/50 hover:text-red-400 transition-colors cursor-pointer"
-    onclick={() => deleteDownload(nodeId, modelId)}
-    title="Delete from this node"
-  >
-    {@render trashIcon()}
-  </button>
-{/snippet}
-
 <div class="min-h-screen bg-exo-dark-gray text-white">
   <HeaderNav showHome={true} />
   <div class="max-w-7xl mx-auto px-4 lg:px-8 py-6 space-y-6">
@@ -540,7 +486,27 @@
                         <span class="text-xs text-white/70"
                           >{formatBytes(cell.totalBytes)}</span
                         >
-                        {@render deleteButton(col.nodeId, row.modelId)}
+                        <button
+                          type="button"
+                          class="text-white/50 hover:text-red-400 transition-colors mt-0.5 cursor-pointer"
+                          onclick={() =>
+                            deleteDownload(col.nodeId, row.modelId)}
+                          title="Delete from this node"
+                        >
+                          <svg
+                            class="w-5 h-5"
+                            viewBox="0 0 20 20"
+                            fill="none"
+                            stroke="currentColor"
+                            stroke-width="2"
+                          >
+                            <path
+                              d="M4 6h12M8 6V4h4v2m1 0v10a1 1 0 01-1 1H8a1 1 0 01-1-1V6h6"
+                              stroke-linecap="round"
+                              stroke-linejoin="round"
+                            ></path>
+                          </svg>
+                        </button>
                       </div>
                     {:else if cell.kind === "downloading"}
                       <div
@@ -567,18 +533,6 @@
                         <span class="text-[10px] text-white/70"
                           >{formatSpeed(cell.speed)}</span
                         >
-                        <div class="flex gap-1 mt-0.5">
-                          <button
-                            type="button"
-                            class="text-white/50 hover:text-exo-yellow transition-colors cursor-pointer"
-                            onclick={() =>
-                              cancelDownload(col.nodeId, row.modelId)}
-                            title="Pause download"
-                          >
-                            {@render pauseIcon()}
-                          </button>
-                          {@render deleteButton(col.nodeId, row.modelId)}
-                        </div>
                       </div>
                     {:else if cell.kind === "pending"}
                       <div
@@ -604,24 +558,32 @@
                               ).toFixed(1)}%"
                             ></div>
                           </div>
-                          <div class="flex gap-1">
-                            {#if row.shardMetadata}
-                              <button
-                                type="button"
-                                class="text-white/50 hover:text-exo-yellow transition-colors cursor-pointer"
-                                onclick={() =>
-                                  startDownload(col.nodeId, row.shardMetadata!)}
-                                title="Resume download on this node"
+                          {#if row.shardMetadata}
+                            <button
+                              type="button"
+                              class="text-white/50 hover:text-exo-yellow transition-colors cursor-pointer"
+                              onclick={() =>
+                                startDownload(col.nodeId, row.shardMetadata!)}
+                              title="Resume download on this node"
+                            >
+                              <svg
+                                class="w-5 h-5"
+                                viewBox="0 0 20 20"
+                                fill="none"
+                                stroke="currentColor"
+                                stroke-width="2"
                               >
-                                {@render downloadIcon()}
-                              </button>
-                            {:else}
-                              <span class="text-white/50 text-[10px]"
-                                >paused</span
-                              >
-                            {/if}
-                            {@render deleteButton(col.nodeId, row.modelId)}
-                          </div>
+                                <path
+                                  d="M10 3v10m0 0l-3-3m3 3l3-3M3 17h14"
+                                  stroke-linecap="round"
+                                  stroke-linejoin="round"
+                                ></path>
+                              </svg>
+                            </button>
+                          {:else}
+                            <span class="text-white/50 text-[10px]">paused</span
+                            >
+                          {/if}
                         {:else if row.shardMetadata}
                           <button
                             type="button"
@@ -630,7 +592,19 @@
                               startDownload(col.nodeId, row.shardMetadata!)}
                             title="Start download on this node"
                           >
-                            {@render downloadIcon("w-6 h-6")}
+                            <svg
+                              class="w-6 h-6"
+                              viewBox="0 0 20 20"
+                              fill="none"
+                              stroke="currentColor"
+                              stroke-width="2"
+                            >
+                              <path
+                                d="M10 3v10m0 0l-3-3m3 3l3-3M3 17h14"
+                                stroke-linecap="round"
+                                stroke-linejoin="round"
+                              ></path>
+                            </svg>
                           </button>
                         {:else}
                           <span class="text-white/40 text-sm">...</span>
@@ -652,20 +626,29 @@
                             clip-rule="evenodd"
                           ></path>
                         </svg>
-                        <div class="flex gap-1">
-                          {#if row.shardMetadata}
-                            <button
-                              type="button"
-                              class="text-white/50 hover:text-exo-yellow transition-colors cursor-pointer"
-                              onclick={() =>
-                                startDownload(col.nodeId, row.shardMetadata!)}
-                              title="Retry download on this node"
+                        {#if row.shardMetadata}
+                          <button
+                            type="button"
+                            class="text-white/50 hover:text-exo-yellow transition-colors cursor-pointer"
+                            onclick={() =>
+                              startDownload(col.nodeId, row.shardMetadata!)}
+                            title="Retry download on this node"
+                          >
+                            <svg
+                              class="w-5 h-5"
+                              viewBox="0 0 20 20"
+                              fill="none"
+                              stroke="currentColor"
+                              stroke-width="2"
                             >
-                              {@render downloadIcon()}
-                            </button>
-                          {/if}
-                          {@render deleteButton(col.nodeId, row.modelId)}
-                        </div>
+                              <path
+                                d="M10 3v10m0 0l-3-3m3 3l3-3M3 17h14"
+                                stroke-linecap="round"
+                                stroke-linejoin="round"
+                              ></path>
+                            </svg>
+                          </button>
+                        {/if}
                       </div>
                     {:else}
                       <div
@@ -683,7 +666,19 @@
                               startDownload(col.nodeId, row.shardMetadata!)}
                             title="Download to this node"
                           >
-                            {@render downloadIcon()}
+                            <svg
+                              class="w-5 h-5"
+                              viewBox="0 0 20 20"
+                              fill="none"
+                              stroke="currentColor"
+                              stroke-width="2"
+                            >
+                              <path
+                                d="M10 3v10m0 0l-3-3m3 3l3-3M3 17h14"
+                                stroke-linecap="round"
+                                stroke-linejoin="round"
+                              ></path>
+                            </svg>
                           </button>
                         {/if}
                       </div>

src/exo/api/main.py

@@ -55,8 +55,6 @@ from exo.api.types import (
     BenchImageGenerationResponse,
     BenchImageGenerationTaskParams,
     CancelCommandResponse,
-    CancelDownloadParams,
-    CancelDownloadResponse,
     ChatCompletionChoice,
     ChatCompletionMessage,
     ChatCompletionRequest,
@@ -149,7 +147,6 @@ from exo.shared.types.chunks import (
 )
 from exo.shared.types.commands import (
     AddCustomModelCard,
-    CancelDownload,
     Command,
     CreateInstance,
     DeleteCustomModelCard,
@@ -354,7 +351,6 @@ class API:
         self.app.get("/events")(self.stream_events)
         self.app.post("/download/start")(self.start_download)
         self.app.delete("/download/{node_id}/{model_id:path}")(self.delete_download)
-        self.app.post("/download/cancel")(self.cancel_download)
         self.app.get("/v1/traces")(self.list_traces)
         self.app.post("/v1/traces/delete")(self.delete_traces)
         self.app.get("/v1/traces/{task_id}")(self.get_trace)
@@ -1883,17 +1879,6 @@ class API:
         await self._send_download(command)
         return DeleteDownloadResponse(command_id=command.command_id)
 
-    async def cancel_download(
-        self,
-        payload: CancelDownloadParams,
-    ) -> CancelDownloadResponse:
-        command = CancelDownload(
-            target_node_id=payload.target_node_id,
-            model_id=payload.model_id,
-        )
-        await self._send_download(command)
-        return CancelDownloadResponse(command_id=command.command_id)
-
     @staticmethod
     def _get_trace_path(task_id: str) -> Path:
         trace_path = EXO_TRACING_CACHE_DIR / f"trace_{task_id}.json"

src/exo/api/types/__init__.py

@@ -5,8 +5,6 @@ from .api import BenchChatCompletionResponse as BenchChatCompletionResponse
 from .api import BenchImageGenerationResponse as BenchImageGenerationResponse
 from .api import BenchImageGenerationTaskParams as BenchImageGenerationTaskParams
 from .api import CancelCommandResponse as CancelCommandResponse
-from .api import CancelDownloadParams as CancelDownloadParams
-from .api import CancelDownloadResponse as CancelDownloadResponse
 from .api import ChatCompletionChoice as ChatCompletionChoice
 from .api import ChatCompletionContentPart as ChatCompletionContentPart
 from .api import ChatCompletionMessage as ChatCompletionMessage

src/exo/api/types/api.py

@@ -430,15 +430,6 @@ class DeleteDownloadResponse(CamelCaseModel):
     command_id: CommandId
 
 
-class CancelDownloadParams(CamelCaseModel):
-    target_node_id: NodeId
-    model_id: ModelId
-
-
-class CancelDownloadResponse(CamelCaseModel):
-    command_id: CommandId
-
-
 class TraceEventResponse(CamelCaseModel):
     name: str
     start_us: int

src/exo/download/coordinator.py

@@ -158,17 +158,10 @@ class DownloadCoordinator:
             logger.info(f"Cancelling download for {model_id}")
             self.active_downloads[model_id].cancel()
             current_status = self.download_status[model_id]
-            downloaded = Memory()
-            total = Memory()
-            if isinstance(current_status, DownloadOngoing):
-                downloaded = current_status.download_progress.downloaded
-                total = current_status.download_progress.total
             pending = DownloadPending(
                 shard_metadata=current_status.shard_metadata,
                 node_id=self.node_id,
                 model_directory=self._default_model_dir(model_id),
-                downloaded=downloaded,
-                total=total,
             )
             self.download_status[model_id] = pending
             await self.event_sender.send(

src/exo/download/tests/test_cancel_download.py

@@ -1,309 +0,0 @@
-"""Tests for cancelling (pausing) an active download via CancelDownload command."""
-
-import asyncio
-import contextlib
-from collections.abc import AsyncIterator, Awaitable
-from datetime import timedelta
-from pathlib import Path
-from typing import Callable
-
-from exo.download.coordinator import DownloadCoordinator
-from exo.download.download_utils import RepoDownloadProgress
-from exo.download.impl_shard_downloader import SingletonShardDownloader
-from exo.download.shard_downloader import ShardDownloader
-from exo.shared.models.model_cards import ModelCard, ModelId, ModelTask
-from exo.shared.types.commands import (
-    CancelDownload,
-    ForwarderDownloadCommand,
-    StartDownload,
-)
-from exo.shared.types.common import NodeId, SystemId
-from exo.shared.types.events import Event, NodeDownloadProgress
-from exo.shared.types.memory import Memory
-from exo.shared.types.worker.downloads import DownloadPending
-from exo.shared.types.worker.shards import PipelineShardMetadata, ShardMetadata
-from exo.utils.channels import Receiver, Sender, channel
-
-NODE_ID = NodeId("aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa")
-MODEL_ID = ModelId("test-org/test-model")
-
-
-def _make_shard(model_id: ModelId = MODEL_ID) -> ShardMetadata:
-    return PipelineShardMetadata(
-        model_card=ModelCard(
-            model_id=model_id,
-            storage_size=Memory.from_mb(100),
-            n_layers=28,
-            hidden_size=1024,
-            supports_tensor=False,
-            tasks=[ModelTask.TextGeneration],
-        ),
-        device_rank=0,
-        world_size=1,
-        start_layer=0,
-        end_layer=28,
-        n_layers=28,
-    )
-
-
-class SlowShardDownloader(ShardDownloader):
-    """Fake downloader that blocks during ensure_shard until cancelled,
-    simulating a long-running download."""
-
-    def __init__(self) -> None:
-        self._progress_callbacks: list[
-            Callable[[ShardMetadata, RepoDownloadProgress], Awaitable[None]]
-        ] = []
-        self.download_started = asyncio.Event()
-
-    def on_progress(
-        self,
-        callback: Callable[[ShardMetadata, RepoDownloadProgress], Awaitable[None]],
-    ) -> None:
-        self._progress_callbacks.append(callback)
-
-    async def ensure_shard(
-        self,
-        shard: ShardMetadata,
-        config_only: bool = False,  # noqa: ARG002
-    ) -> Path:
-        # Fire an in-progress callback, then block forever (until cancelled)
-        progress = RepoDownloadProgress(
-            repo_id=str(shard.model_card.model_id),
-            repo_revision="main",
-            shard=shard,
-            completed_files=0,
-            total_files=1,
-            downloaded=Memory.from_mb(50),
-            downloaded_this_session=Memory.from_mb(50),
-            total=Memory.from_mb(100),
-            overall_speed=1024 * 1024,
-            overall_eta=timedelta(seconds=50),
-            status="in_progress",
-        )
-        for cb in self._progress_callbacks:
-            await cb(shard, progress)
-        self.download_started.set()
-        # Block until cancelled
-        await asyncio.Event().wait()
-        return (
-            Path("/fake/models") / shard.model_card.model_id.normalize()
-        )  # pragma: no cover
-
-    async def get_shard_download_status(
-        self,
-    ) -> AsyncIterator[tuple[Path, RepoDownloadProgress]]:
-        if False:  # noqa: SIM108  # empty async generator
-            yield (
-                Path(),
-                RepoDownloadProgress(  # pyright: ignore[reportUnreachable]
-                    repo_id="",
-                    repo_revision="",
-                    shard=_make_shard(),
-                    completed_files=0,
-                    total_files=0,
-                    downloaded=Memory.from_bytes(0),
-                    downloaded_this_session=Memory.from_bytes(0),
-                    total=Memory.from_bytes(0),
-                    overall_speed=0,
-                    overall_eta=timedelta(seconds=0),
-                    status="not_started",
-                ),
-            )
-
-    async def get_shard_download_status_for_shard(
-        self,
-        shard: ShardMetadata,
-    ) -> RepoDownloadProgress:
-        return RepoDownloadProgress(
-            repo_id=str(shard.model_card.model_id),
-            repo_revision="main",
-            shard=shard,
-            completed_files=0,
-            total_files=1,
-            downloaded=Memory.from_bytes(0),
-            downloaded_this_session=Memory.from_bytes(0),
-            total=Memory.from_mb(100),
-            overall_speed=0,
-            overall_eta=timedelta(seconds=0),
-            status="not_started",
-        )
-
-
-def _setup_coordinator(
-    downloader: ShardDownloader,
-) -> tuple[
-    DownloadCoordinator,
-    Sender[ForwarderDownloadCommand],
-    Receiver[Event],
-]:
-    cmd_send, cmd_recv = channel[ForwarderDownloadCommand]()
-    event_send, event_recv = channel[Event]()
-    wrapped = SingletonShardDownloader(downloader)
-    coordinator = DownloadCoordinator(
-        node_id=NODE_ID,
-        shard_downloader=wrapped,
-        download_command_receiver=cmd_recv,
-        event_sender=event_send,
-    )
-    return coordinator, cmd_send, event_recv
-
-
-async def _wait_for_pending(
-    event_recv: Receiver[Event], model_id: ModelId, timeout: float = 2.0
-) -> DownloadPending | None:
-    """Drain events until we see a DownloadPending for the given model, or timeout."""
-    try:
-        async with asyncio.timeout(timeout):
-            while True:
-                event = await event_recv.receive()
-                if (
-                    isinstance(event, NodeDownloadProgress)
-                    and isinstance(event.download_progress, DownloadPending)
-                    and event.download_progress.shard_metadata.model_card.model_id
-                    == model_id
-                ):
-                    return event.download_progress
-    except TimeoutError:
-        return None
-
-
-async def test_cancel_active_download_transitions_to_pending() -> None:
-    """Cancelling an in-progress download should emit a DownloadPending event
-    and remove the model from active_downloads."""
-    slow_downloader = SlowShardDownloader()
-    coordinator, cmd_send, event_recv = _setup_coordinator(slow_downloader)
-    shard = _make_shard()
-    origin = SystemId("test")
-
-    coordinator_task = asyncio.create_task(coordinator.run())
-    try:
-        # Start a download
-        await cmd_send.send(
-            ForwarderDownloadCommand(
-                origin=origin,
-                command=StartDownload(target_node_id=NODE_ID, shard_metadata=shard),
-            )
-        )
-
-        # Wait for the download to actually start (blocking in ensure_shard)
-        await asyncio.wait_for(slow_downloader.download_started.wait(), timeout=2.0)
-
-        # Drain any events emitted before the cancel (initial DownloadPending, DownloadOngoing)
-        while True:
-            try:
-                async with asyncio.timeout(0.1):
-                    await event_recv.receive()
-            except TimeoutError:
-                break
-
-        # Cancel the download
-        await cmd_send.send(
-            ForwarderDownloadCommand(
-                origin=origin,
-                command=CancelDownload(target_node_id=NODE_ID, model_id=MODEL_ID),
-            )
-        )
-
-        # Should receive a DownloadPending event with preserved progress
-        pending = await _wait_for_pending(event_recv, MODEL_ID)
-        assert pending is not None, "Cancel should emit DownloadPending"
-        assert pending.shard_metadata.model_card.model_id == MODEL_ID
-        assert pending.total == Memory.from_mb(100), "Should preserve total bytes"
-
-        # Give coordinator time to clean up
-        await asyncio.sleep(0.05)
-
-        # Model should no longer be in active_downloads
-        assert MODEL_ID not in coordinator.active_downloads
-        # But should still be in download_status as pending
-        assert MODEL_ID in coordinator.download_status
-        assert isinstance(coordinator.download_status[MODEL_ID], DownloadPending)
-    finally:
-        await coordinator.shutdown()
-        coordinator_task.cancel()
-        with contextlib.suppress(asyncio.CancelledError):
-            await coordinator_task
-
-
-async def test_cancel_nonexistent_download_is_noop() -> None:
-    """Cancelling a model that isn't being downloaded should be a no-op."""
-    slow_downloader = SlowShardDownloader()
-    coordinator, cmd_send, event_recv = _setup_coordinator(slow_downloader)
-    origin = SystemId("test")
-
-    coordinator_task = asyncio.create_task(coordinator.run())
-    try:
-        # Cancel a model that was never started
-        await cmd_send.send(
-            ForwarderDownloadCommand(
-                origin=origin,
-                command=CancelDownload(target_node_id=NODE_ID, model_id=MODEL_ID),
-            )
-        )
-
-        # Should NOT receive any DownloadPending event
-        pending = await _wait_for_pending(event_recv, MODEL_ID, timeout=0.5)
-        assert pending is None, "Cancel of non-existent download should not emit events"
-
-        # Coordinator state should be empty
-        assert MODEL_ID not in coordinator.active_downloads
-        assert MODEL_ID not in coordinator.download_status
-    finally:
-        await coordinator.shutdown()
-        coordinator_task.cancel()
-        with contextlib.suppress(asyncio.CancelledError):
-            await coordinator_task
-
-
-async def test_cancel_then_resume_download() -> None:
-    """After cancelling, re-issuing StartDownload should restart the download."""
-    slow_downloader = SlowShardDownloader()
-    coordinator, cmd_send, event_recv = _setup_coordinator(slow_downloader)
-    shard = _make_shard()
-    origin = SystemId("test")
-
-    coordinator_task = asyncio.create_task(coordinator.run())
-    try:
-        # Start download
-        await cmd_send.send(
-            ForwarderDownloadCommand(
-                origin=origin,
-                command=StartDownload(target_node_id=NODE_ID, shard_metadata=shard),
-            )
-        )
-        await asyncio.wait_for(slow_downloader.download_started.wait(), timeout=2.0)
-
-        # Cancel
-        await cmd_send.send(
-            ForwarderDownloadCommand(
-                origin=origin,
-                command=CancelDownload(target_node_id=NODE_ID, model_id=MODEL_ID),
-            )
-        )
-        pending = await _wait_for_pending(event_recv, MODEL_ID)
-        assert pending is not None, "Cancel should emit DownloadPending"
-
-        await asyncio.sleep(0.05)
-
-        # Reset the event so we can detect the next download start
-        slow_downloader.download_started.clear()
-
-        # Resume by sending StartDownload again
-        await cmd_send.send(
-            ForwarderDownloadCommand(
-                origin=origin,
-                command=StartDownload(target_node_id=NODE_ID, shard_metadata=shard),
-            )
-        )
-
-        # The download should restart
-        await asyncio.wait_for(slow_downloader.download_started.wait(), timeout=2.0)
-        assert MODEL_ID in coordinator.active_downloads, (
-            "Model should be actively downloading again after resume"
-        )
-    finally:
-        await coordinator.shutdown()
-        coordinator_task.cancel()
-        with contextlib.suppress(asyncio.CancelledError):
-            await coordinator_task

src/exo/shared/constants.py

@@ -97,3 +97,5 @@ EXO_OFFLINE = os.getenv("EXO_OFFLINE", "false").lower() == "true"
 EXO_TRACING_ENABLED = os.getenv("EXO_TRACING_ENABLED", "false").lower() == "true"
 
 EXO_MAX_CONCURRENT_REQUESTS = int(os.getenv("EXO_MAX_CONCURRENT_REQUESTS", "8"))
+
+EXO_MAX_INSTANCE_RETRIES = 3

src/exo/utils/keyed_backoff.py

@@ -1,10 +1,9 @@
 import time
-from typing import Generic, TypeVar
-
-K = TypeVar("K")
+from typing import final
 
 
-class KeyedBackoff(Generic[K]):
+@final
+class KeyedBackoff[K]:
     """Tracks exponential backoff state per key."""
 
     def __init__(self, base: float = 0.5, cap: float = 10.0):
@@ -26,6 +25,10 @@ class KeyedBackoff(Generic[K]):
         self._last_time[key] = time.monotonic()
         self._attempts[key] = self._attempts.get(key, 0) + 1
 
+    def attempts(self, key: K) -> int:
+        """Return the number of recorded attempts for a key."""
+        return self._attempts.get(key, 0)
+
     def reset(self, key: K) -> None:
         """Reset backoff state for a key (e.g., on success)."""
         self._attempts.pop(key, None)

src/exo/worker/main.py

@@ -9,9 +9,11 @@ from loguru import logger
 from exo.api.types import ImageEditsTaskParams
 from exo.download.download_utils import is_read_only_model_dir, resolve_existing_model
 from exo.shared.apply import apply
+from exo.shared.constants import EXO_MAX_INSTANCE_RETRIES
 from exo.shared.models.model_cards import ModelId, add_to_card_cache, delete_custom_card
 from exo.shared.types.chunks import InputImageChunk
 from exo.shared.types.commands import (
+    DeleteInstance,
     ForwarderCommand,
     ForwarderDownloadCommand,
     StartDownload,
@@ -23,6 +25,7 @@ from exo.shared.types.events import (
     Event,
     IndexedEvent,
     InputChunkReceived,
+    InstanceDeleted,
     NodeDownloadProgress,
     NodeGatheredInfo,
     TaskCreated,
@@ -44,6 +47,7 @@ from exo.shared.types.tasks import (
 )
 from exo.shared.types.topology import Connection, SocketConnection
 from exo.shared.types.worker.downloads import DownloadCompleted
+from exo.shared.types.worker.instances import InstanceId
 from exo.shared.types.worker.runners import RunnerId
 from exo.utils.channels import Receiver, Sender, channel
 from exo.utils.info_gatherer.info_gatherer import GatheredInfo, InfoGatherer
@@ -84,6 +88,9 @@ class Worker:
         self.image_cache: dict[str, str] = {}
 
         self._download_backoff: KeyedBackoff[ModelId] = KeyedBackoff(base=0.5, cap=10.0)
+        self._instance_backoff: KeyedBackoff[InstanceId] = KeyedBackoff(
+            base=0.5, cap=10.0
+        )
         self._stopped: anyio.Event = anyio.Event()
 
     async def run(self):
@@ -127,6 +134,9 @@ class Worker:
                 self.state = apply(self.state, event=event)
                 event = event.event
 
+                if isinstance(event, InstanceDeleted):
+                    self._instance_backoff.reset(event.instance_id)
+
                 # Buffer input image chunks for image editing
                 if isinstance(event, InputChunkReceived):
                     cmd_id = event.command_id
@@ -156,15 +166,24 @@ class Worker:
                 self.state.runners,
                 self.state.tasks,
                 self.input_chunk_buffer,
+                self._instance_backoff,
+                self._download_backoff,
             )
             if task is None:
                 continue
 
-            # Gate DownloadModel on backoff BEFORE emitting TaskCreated
-            # to prevent flooding the event log with useless events
-            if isinstance(task, DownloadModel):
-                model_id = task.shard_metadata.model_card.model_id
-                if not self._download_backoff.should_proceed(model_id):
+            if isinstance(task, CreateRunner):
+                iid = task.instance_id
+                if self._instance_backoff.attempts(iid) >= EXO_MAX_INSTANCE_RETRIES:
+                    logger.warning(
+                        f"Instance {iid} exceeded {EXO_MAX_INSTANCE_RETRIES} retries, requesting deletion"
+                    )
+                    await self.command_sender.send(
+                        ForwarderCommand(
+                            origin=self._system_id,
+                            command=DeleteInstance(instance_id=iid),
+                        )
+                    )
                     continue
 
             logger.info(f"Worker plan: {task.__class__.__name__}")
@@ -175,6 +194,7 @@ class Worker:
             match task:
                 case CreateRunner():
                     self._create_supervisor(task)
+                    self._instance_backoff.record_attempt(task.instance_id)
                     await self.event_sender.send(
                         TaskStatusUpdated(
                             task_id=task.task_id, task_status=TaskStatus.Complete

src/exo/worker/plan.py

@@ -3,7 +3,7 @@
 from collections.abc import Mapping, Sequence
 
 from exo.shared.types.chunks import InputImageChunk
-from exo.shared.types.common import CommandId, NodeId
+from exo.shared.types.common import CommandId, ModelId, NodeId
 from exo.shared.types.tasks import (
     CancelTask,
     ConnectToGroup,
@@ -39,6 +39,7 @@ from exo.shared.types.worker.runners import (
     RunnerStatus,
     RunnerWarmingUp,
 )
+from exo.utils.keyed_backoff import KeyedBackoff
 from exo.worker.runner.runner_supervisor import RunnerSupervisor
 
 
@@ -50,18 +51,22 @@ def plan(
     instances: Mapping[InstanceId, Instance],
     all_runners: Mapping[RunnerId, RunnerStatus],  # all global
     tasks: Mapping[TaskId, Task],
-    input_chunk_buffer: Mapping[CommandId, Mapping[int, InputImageChunk]] | None = None,
+    input_chunk_buffer: Mapping[CommandId, Mapping[int, InputImageChunk]],
+    instance_backoff: KeyedBackoff[InstanceId],
+    download_backoff: KeyedBackoff[ModelId],
 ) -> Task | None:
     # Python short circuiting OR logic should evaluate these sequentially.
     return (
         _cancel_tasks(runners, tasks)
         or _kill_runner(runners, all_runners, instances)
-        or _create_runner(node_id, runners, instances)
-        or _model_needs_download(node_id, runners, global_download_status)
+        or _create_runner(node_id, runners, instances, instance_backoff)
+        or _model_needs_download(
+            node_id, runners, global_download_status, download_backoff
+        )
         or _init_distributed_backend(runners, all_runners)
         or _load_model(runners, all_runners, global_download_status)
         or _ready_to_warmup(runners, all_runners)
-        or _pending_tasks(runners, tasks, all_runners, input_chunk_buffer or {})
+        or _pending_tasks(runners, tasks, all_runners, input_chunk_buffer)
     )
 
 
@@ -92,8 +97,12 @@ def _create_runner(
     node_id: NodeId,
     runners: Mapping[RunnerId, RunnerSupervisor],
     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
@@ -116,6 +125,7 @@ def _model_needs_download(
     node_id: NodeId,
     runners: Mapping[RunnerId, RunnerSupervisor],
     global_download_status: Mapping[NodeId, Sequence[DownloadProgress]],
+    download_backoff: KeyedBackoff[ModelId],
 ) -> DownloadModel | None:
     local_downloads = global_download_status.get(node_id, [])
     download_status = {
@@ -124,12 +134,16 @@ def _model_needs_download(
 
     for runner in runners.values():
         model_id = runner.bound_instance.bound_shard.model_card.model_id
-        if isinstance(runner.status, RunnerIdle) and (
-            model_id not in download_status
-            or not isinstance(
-                download_status[model_id],
-                (DownloadOngoing, DownloadCompleted, DownloadFailed),
+        if (
+            isinstance(runner.status, RunnerIdle)
+            and (
+                model_id not in download_status
+                or not isinstance(
+                    download_status[model_id],
+                    (DownloadOngoing, DownloadCompleted, DownloadFailed),
+                )
             )
+            and download_backoff.should_proceed(model_id)
         ):
             # We don't invalidate download_status randomly in case a file gets deleted on disk
             return DownloadModel(
@@ -272,7 +286,7 @@ def _pending_tasks(
     runners: Mapping[RunnerId, RunnerSupervisor],
     tasks: Mapping[TaskId, Task],
     all_runners: Mapping[RunnerId, RunnerStatus],
-    input_chunk_buffer: Mapping[CommandId, Mapping[int, InputImageChunk]] | None,
+    input_chunk_buffer: Mapping[CommandId, Mapping[int, InputImageChunk]],
 ) -> Task | None:
     for task in tasks.values():
         # for now, just forward chat completions
@@ -287,7 +301,6 @@ def _pending_tasks(
         if isinstance(task, (ImageEdits, TextGeneration)):
             expected_image_chunks = task.task_params.total_input_chunks
         if expected_image_chunks > 0:
-            assert input_chunk_buffer is not None
             cmd_id = task.command_id
             received = len(input_chunk_buffer.get(cmd_id, {}))
             if received < expected_image_chunks:

src/exo/worker/tests/unittests/test_plan/test_download_and_loading.py

@@ -8,6 +8,7 @@ from exo.shared.types.worker.runners import (
     RunnerConnected,
     RunnerIdle,
 )
+from exo.utils.keyed_backoff import KeyedBackoff
 from exo.worker.tests.constants import (
     INSTANCE_1_ID,
     MODEL_A_ID,
@@ -52,6 +53,9 @@ def test_plan_requests_download_when_waiting_and_shard_not_downloaded():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert isinstance(result, plan_mod.DownloadModel)
@@ -104,6 +108,9 @@ def test_plan_loads_model_when_all_shards_downloaded_and_waiting():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert isinstance(result, LoadModel)
@@ -146,6 +153,9 @@ def test_plan_does_not_request_download_when_shard_already_downloaded():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert not isinstance(result, plan_mod.DownloadModel)
@@ -193,6 +203,9 @@ def test_plan_does_not_load_model_until_all_shards_downloaded_globally():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert result is None
@@ -213,6 +226,9 @@ def test_plan_does_not_load_model_until_all_shards_downloaded_globally():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert result is not None

src/exo/worker/tests/unittests/test_plan/test_runner_lifecycle.py

@@ -9,6 +9,7 @@ from exo.shared.types.worker.runners import (
     RunnerReady,
     RunnerStatus,
 )
+from exo.utils.keyed_backoff import KeyedBackoff
 from exo.worker.tests.constants import (
     INSTANCE_1_ID,
     MODEL_A_ID,
@@ -52,6 +53,9 @@ def test_plan_kills_runner_when_instance_missing():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert isinstance(result, Shutdown)
@@ -91,6 +95,9 @@ def test_plan_kills_runner_when_sibling_failed():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert isinstance(result, Shutdown)
@@ -122,6 +129,9 @@ def test_plan_creates_runner_when_missing_for_node():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     # We patched plan_mod.CreateRunner → CreateRunner
@@ -160,6 +170,9 @@ def test_plan_does_not_create_runner_when_supervisor_already_present():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert result is None
@@ -189,6 +202,9 @@ def test_plan_does_not_create_runner_for_unassigned_node():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert result is None

src/exo/worker/tests/unittests/test_plan/test_task_forwarding.py

@@ -9,6 +9,7 @@ from exo.shared.types.worker.runners import (
     RunnerReady,
     RunnerRunning,
 )
+from exo.utils.keyed_backoff import KeyedBackoff
 from exo.worker.tests.constants import (
     COMMAND_1_ID,
     INSTANCE_1_ID,
@@ -71,6 +72,9 @@ def test_plan_forwards_pending_chat_completion_when_runner_ready():
         instances=instances,
         all_runners=all_runners,
         tasks={TASK_1_ID: task},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert result is task
@@ -120,6 +124,9 @@ def test_plan_does_not_forward_chat_completion_if_any_runner_not_ready():
         instances=instances,
         all_runners=all_runners,
         tasks={TASK_1_ID: task},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert result is None
@@ -166,6 +173,9 @@ def test_plan_does_not_forward_tasks_for_other_instances():
         instances=instances,
         all_runners=all_runners,
         tasks={foreign_task.task_id: foreign_task},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert result is None
@@ -230,6 +240,9 @@ def test_plan_ignores_non_pending_or_non_chat_tasks():
         instances=instances,
         all_runners=all_runners,
         tasks={TASK_1_ID: completed_task, other_task_id: other_task},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert result is None
@@ -269,6 +282,9 @@ def test_plan_returns_none_when_nothing_to_do():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert result is None

src/exo/worker/tests/unittests/test_plan/test_warmup.py

@@ -7,6 +7,7 @@ from exo.shared.types.worker.runners import (
     RunnerLoading,
     RunnerWarmingUp,
 )
+from exo.utils.keyed_backoff import KeyedBackoff
 from exo.worker.tests.constants import (
     INSTANCE_1_ID,
     MODEL_A_ID,
@@ -61,6 +62,9 @@ def test_plan_starts_warmup_for_accepting_rank_when_all_loaded_or_warming():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert isinstance(result, StartWarmup)
@@ -102,6 +106,9 @@ def test_plan_starts_warmup_for_rank_zero_after_others_warming():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert isinstance(result, StartWarmup)
@@ -142,6 +149,9 @@ def test_plan_does_not_start_warmup_for_non_zero_rank_until_all_loaded_or_warmin
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert result is None
@@ -186,6 +196,9 @@ def test_plan_does_not_start_warmup_for_rank_zero_until_others_warming():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert result is None
@@ -202,6 +215,9 @@ def test_plan_does_not_start_warmup_for_rank_zero_until_others_warming():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert isinstance(result, StartWarmup)
@@ -245,6 +261,9 @@ def test_plan_starts_warmup_for_connecting_rank_after_others_warming():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert isinstance(result, StartWarmup)
@@ -287,6 +306,9 @@ def test_plan_does_not_start_warmup_for_accepting_rank_until_all_loaded_or_warmi
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert result is None
@@ -328,6 +350,9 @@ def test_plan_does_not_start_warmup_for_connecting_rank_until_others_warming():
         instances=instances,
         all_runners=all_runners,
         tasks={},
+        input_chunk_buffer={},
+        instance_backoff=KeyedBackoff(),
+        download_backoff=KeyedBackoff(),
     )
 
     assert result is None