Bound recovered leader connection attempts

Recovered jobs could stop immediately after logging that their state
machine had started. A stale worker address left tonic's connection
future pending forever because it had no connection deadline.

Apply a 10-second timeout to each attempt while retaining the existing
retry and recovery behavior. Exhausted attempts now let the state machine
handle the unavailable leader instead of remaining stuck before its first
state executes.

Author: cmackenzie1

crates/arroyo-controller/src/job_controller/leader_manager.rs

@@ -18,6 +18,8 @@ use tonic::codegen::InterceptedService;
 use tonic::transport::Channel;
 use tracing::{info, warn};
 
+const LEADER_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
+
 pub struct LeaderManager {
     leader_client: JobStatusGrpcClient<InterceptedService<Channel, InjectWorkerId>>,
     pub job_id: JobId,
@@ -35,13 +37,23 @@ impl LeaderManager {
         address: String,
     ) -> anyhow::Result<Self> {
         let leader_client = retry!(
-            job_status_client(
-                "controller",
-                &config().worker.tls,
-                worker_id,
-                address.clone()
+            match tokio::time::timeout(
+                LEADER_CONNECT_TIMEOUT,
+                job_status_client(
+                    "controller",
+                    &config().worker.tls,
+                    worker_id,
+                    address.clone(),
+                ),
             )
-            .await,
+            .await
+            {
+                Ok(result) => result,
+                Err(_) => Err(anyhow!(
+                    "timed out connecting to worker leader after {:?}",
+                    LEADER_CONNECT_TIMEOUT
+                )),
+            },
             5,
             Duration::from_millis(100),
             Duration::from_secs(2),