Apply configurable gRPC connection timeout

The recovered-leader fix bounded only one connection path, while other
controller and worker clients could still wait indefinitely for tonic to
establish a transport connection.

Add a global `grpc.connect-timeout` setting with a 10-second default and
apply it to every endpoint produced by `grpc_channel_builder`. Remove the
leader-specific wrapper while retaining its retry policy, and cover both
the default and environment override configuration.

Author: cmackenzie1

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

@@ -18,8 +18,6 @@ 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,
@@ -37,23 +35,13 @@ impl LeaderManager {
         address: String,
     ) -> anyhow::Result<Self> {
         let leader_client = retry!(
-            match tokio::time::timeout(
-                LEADER_CONNECT_TIMEOUT,
-                job_status_client(
-                    "controller",
-                    &config().worker.tls,
-                    worker_id,
-                    address.clone(),
-                ),
+            job_status_client(
+                "controller",
+                &config().worker.tls,
+                worker_id,
+                address.clone()
             )
-            .await
-            {
-                Ok(result) => result,
-                Err(_) => Err(anyhow!(
-                    "timed out connecting to worker leader after {:?}",
-                    LEADER_CONNECT_TIMEOUT
-                )),
-            },
+            .await,
             5,
             Duration::from_millis(100),
             Duration::from_secs(2),

crates/arroyo-rpc/default.toml

@@ -2,6 +2,9 @@ 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,6 +193,9 @@ 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,
 
@@ -275,6 +278,13 @@ 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 {
@@ -1047,6 +1057,7 @@ impl TlsConfig {
 #[cfg(test)]
 mod tests {
     use crate::config::{Config, DatabaseType, Scheduler, SchemaName, SqliteConfig, load_config};
+    use std::time::Duration;
     use url::Url;
 
     #[test]
@@ -1073,6 +1084,30 @@ 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();
-    if let Some(target_tls) = config.get_tls_config(target_tls) {
+    let endpoint = if let Some(target_tls) = config.get_tls_config(target_tls) {
         let mut endpoint = Url::parse(&endpoint)?;
         endpoint
             .set_scheme("https")
@@ -1031,11 +1031,13 @@ pub async fn grpc_channel_builder(
             config_builder = config_builder.identity(Identity::from_pem(our_tls.cert, our_tls.key));
         }
 
-        Ok(b.tls_config(config_builder).context("configuring TLS")?)
+        b.tls_config(config_builder).context("configuring TLS")?
     } else {
         debug!("connecting to grpc endpoint {endpoint}");
-        Ok(Channel::from_shared(endpoint.to_string())?)
-    }
+        Channel::from_shared(endpoint.to_string())?
+    };
+
+    Ok(endpoint.connect_timeout(*config.grpc.connect_timeout))
 }
 
 /// Connect to a gRPC service with optional TLS