fix: raise AioRpcError on UNAVAILABLE timeout and assert router connection in test

Make UNAVAILABLE timeout in handle_async raise instead of returning silently,
matching the FAILED_PRECONDITION timeout behavior. Add assertion that
connect_router_stream is called after successful UNAVAILABLE retry.

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

@@ -237,9 +237,7 @@ async def _acquire(self):
                             # Old controllers (pre-918d6341) mark offline-but-matching
                             # exporters as Unsatisfiable with reason "NoExporter".
                             # This is transient — retry with a new lease.
-                            if condition_present_and_equal(
-                                result.conditions, "Unsatisfiable", "True", "NoExporter"
-                            ):
+                            if condition_present_and_equal(result.conditions, "Unsatisfiable", "True", "NoExporter"):
                                 await self._handle_no_exporter_retry(spinner, message)
                                 continue
                             logger.debug("Lease %s cannot be satisfied: %s", self.name, message)
@@ -325,38 +323,43 @@ async def handle_async(self, stream):
                     if remaining <= 0:
                         logger.debug(
                             "Exporter not ready and dial timeout (%.1fs) exceeded after %d attempts",
-                            self.dial_timeout, attempt + 1
+                            self.dial_timeout,
+                            attempt + 1,
                         )
                         raise
-                    delay = min(base_delay * (2 ** attempt), max_delay, remaining)
+                    delay = min(base_delay * (2**attempt), max_delay, remaining)
                     logger.debug(
                         "Exporter not ready, retrying Dial in %.1fs (attempt %d, %.1fs remaining)",
-                        delay, attempt + 1, remaining
+                        delay,
+                        attempt + 1,
+                        remaining,
                     )
                     await sleep(delay)
                     attempt += 1
                     continue
                 if e.code() == grpc.StatusCode.UNAVAILABLE:
                     remaining = deadline - time.monotonic()
                     if remaining <= 0:
-                        logger.warning(
+                        logger.debug(
                             "Exporter unavailable and dial timeout (%.1fs) exceeded after %d attempts",
-                            self.dial_timeout, attempt + 1
+                            self.dial_timeout,
+                            attempt + 1,
                         )
-                        return
-                    delay = min(base_delay * (2 ** attempt), max_delay, remaining)
+                        raise
+                    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
+                        delay,
+                        attempt + 1,
+                        remaining,
                     )
                     await sleep(delay)
                     attempt += 1
                     continue
                 if "permission denied" in str(e.details()).lower():
                     self.lease_transferred = True
                     logger.warning(
-                        "Lease %s has been transferred to another client. "
-                        "Your session is no longer valid.",
+                        "Lease %s has been transferred to another client. Your session is no longer valid.",
                         self.name,
                     )
                 else:

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

@@ -575,11 +575,12 @@ async def mock_dial(request):
 
             await lease.handle_async(stream)
 
-        assert dial_call_count == 2
+            assert dial_call_count == 2
+            mock_connect.assert_called_once_with("endpoint", "token", stream, lease.tls_config, lease.grpc_options)
 
     @pytest.mark.anyio
     async def test_handle_async_unavailable_exceeds_dial_timeout(self):
-        """Dial returns UNAVAILABLE until dial_timeout is exceeded."""
+        """Dial returns UNAVAILABLE until dial_timeout is exceeded, then raises."""
         lease = self._make_lease_for_handle()
         lease.dial_timeout = 0.5
         dial_call_count = 0
@@ -592,6 +593,8 @@ async def mock_dial(request):
         lease.controller.Dial = mock_dial
         stream = Mock()
 
-        await lease.handle_async(stream)
+        with pytest.raises(AioRpcError) as exc_info:
+            await lease.handle_async(stream)
 
+        assert exc_info.value.code() == grpc.StatusCode.UNAVAILABLE
         assert dial_call_count >= 2