fix: retry Dial on transient UNAVAILABLE in Lease.handle_async

When the exporter briefly restarts, the Dial RPC may fail with
UNAVAILABLE. Instead of immediately giving up, retry with exponential
backoff bounded by the existing dial_timeout parameter. This mirrors
the existing FAILED_PRECONDITION retry logic.

Fixes #242

Generated-By: Forge/20260416_202053_681470_11575359_i242
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: raballew

python/packages/jumpstarter/jumpstarter/client/lease.py

@@ -336,7 +336,22 @@ async def handle_async(self, stream):
                     await sleep(delay)
                     attempt += 1
                     continue
-                # Exporter went offline or lease ended - log and exit gracefully
+                if e.code() == grpc.StatusCode.UNAVAILABLE:
+                    remaining = deadline - time.monotonic()
+                    if remaining <= 0:
+                        logger.warning(
+                            "Exporter unavailable and dial timeout (%.1fs) exceeded after %d attempts",
+                            self.dial_timeout, attempt + 1
+                        )
+                        return
+                    delay = min(base_delay * (2 ** attempt), max_delay, remaining)
+                    logger.debug(
+                        "Exporter unavailable, retrying Dial in %.1fs (attempt %d, %.1fs remaining)",
+                        delay, attempt + 1, remaining
+                    )
+                    await sleep(delay)
+                    attempt += 1
+                    continue
                 if "permission denied" in str(e.details()).lower():
                     self.lease_transferred = True
                     logger.warning(

python/packages/jumpstarter/jumpstarter/client/lease_test.py

@@ -4,12 +4,28 @@
 from datetime import datetime, timedelta, timezone
 from unittest.mock import AsyncMock, Mock, patch
 
+import grpc
 import pytest
+from grpc.aio import AioRpcError
 from rich.console import Console
 
 from jumpstarter.client.lease import Lease, LeaseAcquisitionSpinner
 
 
+class MockAioRpcError(AioRpcError):
+    """Mock gRPC error for testing that properly inherits from AioRpcError."""
+
+    def __init__(self, status_code, message=""):
+        self._status_code = status_code
+        self._message = message
+
+    def code(self):
+        return self._status_code
+
+    def details(self):
+        return self._message
+
+
 class TestLeaseAcquisitionSpinner:
     """Test cases for LeaseAcquisitionSpinner class."""
 
@@ -522,3 +538,60 @@ async def get_then_fail():
         callback.assert_called()
         _, remain_arg = callback.call_args[0]
         assert remain_arg == timedelta(0)
+
+
+class TestHandleAsyncUnavailableRetry:
+    """Tests for Lease.handle_async UNAVAILABLE retry behavior."""
+
+    def _make_lease_for_handle(self):
+        lease = object.__new__(Lease)
+        lease.name = "test-lease"
+        lease.dial_timeout = 5.0
+        lease.lease_transferred = False
+        lease.tls_config = Mock()
+        lease.grpc_options = {}
+        lease.controller = Mock()
+        return lease
+
+    @pytest.mark.anyio
+    async def test_handle_async_retries_unavailable_then_succeeds(self):
+        """Dial returns UNAVAILABLE once then succeeds on retry."""
+        lease = self._make_lease_for_handle()
+        dial_call_count = 0
+
+        async def mock_dial(request):
+            nonlocal dial_call_count
+            dial_call_count += 1
+            if dial_call_count == 1:
+                raise MockAioRpcError(grpc.StatusCode.UNAVAILABLE, "temporarily unavailable")
+            return Mock(router_endpoint="endpoint", router_token="token")
+
+        lease.controller.Dial = mock_dial
+
+        with patch("jumpstarter.client.lease.connect_router_stream") as mock_connect:
+            mock_connect.return_value.__aenter__ = AsyncMock()
+            mock_connect.return_value.__aexit__ = AsyncMock(return_value=False)
+            stream = Mock()
+
+            await lease.handle_async(stream)
+
+        assert dial_call_count == 2
+
+    @pytest.mark.anyio
+    async def test_handle_async_unavailable_exceeds_dial_timeout(self):
+        """Dial returns UNAVAILABLE until dial_timeout is exceeded."""
+        lease = self._make_lease_for_handle()
+        lease.dial_timeout = 0.5
+        dial_call_count = 0
+
+        async def mock_dial(request):
+            nonlocal dial_call_count
+            dial_call_count += 1
+            raise MockAioRpcError(grpc.StatusCode.UNAVAILABLE, "permanently unavailable")
+
+        lease.controller.Dial = mock_dial
+        stream = Mock()
+
+        await lease.handle_async(stream)
+
+        assert dial_call_count >= 2