Limit connection timeout to recovered leaders

The global gRPC connection timeout affected every Arroyo service and RPC
client even though the observed stall occurs while reconnecting a recovered
job to its worker leader.

Remove the global configuration and endpoint behavior. Restore the
10-second deadline around each leader connection attempt while preserving
the existing retry and recovery behavior.

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),

crates/arroyo-rpc/default.toml

@@ -2,9 +2,6 @@ checkpoint-url = "/tmp/arroyo/checkpoints"
 default-checkpoint-interval = "10s"
 job-controller = "controller"
 
-[grpc]
-connect-timeout = "10s"
-
 [pipeline]
 source-batch-size = 512
 source-batch-linger = "100ms"

crates/arroyo-rpc/src/config.rs

@@ -193,9 +193,6 @@ fn load_config(paths: &[PathBuf]) -> Figment {
 #[derive(Debug, Deserialize, Serialize, Clone)]
 #[serde(rename_all = "kebab-case", deny_unknown_fields)]
 pub struct Config {
-    /// gRPC client configuration
-    pub grpc: GrpcConfig,
-
     /// API service configuration
     pub api: ApiConfig,
 
@@ -278,13 +275,6 @@ pub struct Config {
     pub disable_telemetry: bool,
 }
 
-#[derive(Debug, Deserialize, Serialize, Clone)]
-#[serde(rename_all = "kebab-case", deny_unknown_fields)]
-pub struct GrpcConfig {
-    /// Maximum time to establish a gRPC connection
-    pub connect_timeout: HumanReadableDuration,
-}
-
 #[derive(Debug, Deserialize, Serialize, Clone, Default)]
 #[serde(rename_all = "kebab-case", deny_unknown_fields)]
 pub enum JobControllerMode {
@@ -1057,7 +1047,6 @@ impl TlsConfig {
 #[cfg(test)]
 mod tests {
     use crate::config::{Config, DatabaseType, Scheduler, SchemaName, SqliteConfig, load_config};
-    use std::time::Duration;
     use url::Url;
 
     #[test]
@@ -1084,30 +1073,6 @@ mod tests {
         }
     }
 
-    #[test]
-    #[allow(clippy::result_large_err)]
-    fn grpc_connect_timeout_defaults_to_ten_seconds() {
-        figment::Jail::expect_with(|_| {
-            let config: Config = load_config(&[]).extract().unwrap();
-
-            assert_eq!(*config.grpc.connect_timeout, Duration::from_secs(10));
-            Ok(())
-        });
-    }
-
-    #[test]
-    #[allow(clippy::result_large_err)]
-    fn grpc_connect_timeout_can_be_overridden_with_environment() {
-        figment::Jail::expect_with(|jail| {
-            jail.set_env("ARROYO__GRPC__CONNECT_TIMEOUT", "3s");
-
-            let config: Config = load_config(&[]).extract().unwrap();
-
-            assert_eq!(*config.grpc.connect_timeout, Duration::from_secs(3));
-            Ok(())
-        });
-    }
-
     #[test]
     #[allow(clippy::result_large_err)]
     fn test_config() {

crates/arroyo-rpc/src/lib.rs

@@ -1004,7 +1004,7 @@ pub async fn grpc_channel_builder(
     target_tls: &Option<TlsConfig>,
 ) -> Result<Endpoint> {
     let config = config();
-    let endpoint = if let Some(target_tls) = config.get_tls_config(target_tls) {
+    if let Some(target_tls) = config.get_tls_config(target_tls) {
         let mut endpoint = Url::parse(&endpoint)?;
         endpoint
             .set_scheme("https")
@@ -1031,13 +1031,11 @@ pub async fn grpc_channel_builder(
             config_builder = config_builder.identity(Identity::from_pem(our_tls.cert, our_tls.key));
         }
 
-        b.tls_config(config_builder).context("configuring TLS")?
+        Ok(b.tls_config(config_builder).context("configuring TLS")?)
     } else {
         debug!("connecting to grpc endpoint {endpoint}");
-        Channel::from_shared(endpoint.to_string())?
-    };
-
-    Ok(endpoint.connect_timeout(*config.grpc.connect_timeout))
+        Ok(Channel::from_shared(endpoint.to_string())?)
+    }
 }
 
 /// Connect to a gRPC service with optional TLS