diff --git a/crates/dbx-core/src/connection.rs b/crates/dbx-core/src/connection.rs index 1ffdd4ff74..1018884bdc 100644 --- a/crates/dbx-core/src/connection.rs +++ b/crates/dbx-core/src/connection.rs @@ -809,7 +809,7 @@ impl AppState { ) } else if db_config.uses_redis_sentinel() { db::redis_driver::RedisConnection::Direct(tokio::sync::Mutex::new( - db::redis_driver::connect_sentinel(&db_config).await?, + self.connect_redis_sentinel(connection_id, &db_config).await?, )) } else { db::redis_driver::RedisConnection::Direct(tokio::sync::Mutex::new( @@ -1093,6 +1093,129 @@ impl AppState { Ok(("127.0.0.1".to_string(), local_port)) } + pub async fn connect_redis_sentinel( + &self, + connection_id: &str, + config: &ConnectionConfig, + ) -> Result { + let transport_layers = config.effective_transport_layers(); + if transport_layers.is_empty() { + return db::redis_driver::connect_sentinel(config).await; + } + + let result = async { + let sentinel_nodes = db::redis_driver::redis_sentinel_node_endpoints(config)?; + let connect_timeout = std::time::Duration::from_secs(config.effective_connect_timeout_secs()); + let layer_count = transport_layers.len(); + let mut last_error = None; + + for sentinel in sentinel_nodes { + let sentinel_tunnel_id = redis_sentinel_transport_id(connection_id, "sentinel", &sentinel); + let sentinel_local_port = match db::transport_layer_tunnel::start_transport_layers( + &sentinel_tunnel_id, + &transport_layers, + &sentinel.host, + sentinel.port, + &self.tunnels, + &self.proxy_tunnels, + ) + .await + { + Ok(port) => port, + Err(err) => { + last_error = + Some(format!("Redis Sentinel {}:{} transport failed: {err}", sentinel.host, sentinel.port)); + continue; + } + }; + + let master = + match db::redis_driver::discover_sentinel_master(config, "127.0.0.1", sentinel_local_port).await { + Ok(master) => master, + Err(err) => { + last_error = Some(format!( + "Redis Sentinel {}:{} master lookup failed: {err}", + sentinel.host, sentinel.port + )); + db::transport_layer_tunnel::stop_transport_layers( + &sentinel_tunnel_id, + layer_count, + &self.tunnels, + &self.proxy_tunnels, + ) + .await; + continue; + } + }; + + let master_tunnel_id = redis_sentinel_transport_id(connection_id, "master", &master); + let master_local_port = match db::transport_layer_tunnel::start_transport_layers( + &master_tunnel_id, + &transport_layers, + &master.host, + master.port, + &self.tunnels, + &self.proxy_tunnels, + ) + .await + { + Ok(port) => port, + Err(err) => { + last_error = Some(format!( + "Redis Sentinel master {}:{} transport failed: {err}", + master.host, master.port + )); + db::transport_layer_tunnel::stop_transport_layers( + &sentinel_tunnel_id, + layer_count, + &self.tunnels, + &self.proxy_tunnels, + ) + .await; + continue; + } + }; + + match db::redis_driver::connect_standalone(config, "127.0.0.1", master_local_port, connect_timeout) + .await + { + Ok(con) => return Ok(con), + Err(err) => { + last_error = Some(format!( + "Redis Sentinel master {}:{} connection failed: {err}", + master.host, master.port + )); + db::transport_layer_tunnel::stop_transport_layers( + &master_tunnel_id, + layer_count, + &self.tunnels, + &self.proxy_tunnels, + ) + .await; + db::transport_layer_tunnel::stop_transport_layers( + &sentinel_tunnel_id, + layer_count, + &self.tunnels, + &self.proxy_tunnels, + ) + .await; + } + } + } + + Err(last_error.unwrap_or_else(|| "Redis Sentinel master discovery failed".to_string())) + } + .await; + + if result.is_err() { + let redis_sentinel_prefix = redis_sentinel_transport_prefix(connection_id); + self.tunnels.stop_tunnels_with_prefix(&redis_sentinel_prefix).await; + self.proxy_tunnels.stop_tunnels_with_prefix(&redis_sentinel_prefix).await; + } + + result + } + pub async fn connect_redis_cluster( &self, connection_id: &str, @@ -1584,6 +1707,9 @@ impl AppState { let redis_cluster_prefix = redis_cluster_transport_prefix(connection_id); self.tunnels.stop_tunnels_with_prefix(&redis_cluster_prefix).await; self.proxy_tunnels.stop_tunnels_with_prefix(&redis_cluster_prefix).await; + let redis_sentinel_prefix = redis_sentinel_transport_prefix(connection_id); + self.tunnels.stop_tunnels_with_prefix(&redis_sentinel_prefix).await; + self.proxy_tunnels.stop_tunnels_with_prefix(&redis_sentinel_prefix).await; db::transport_layer_tunnel::stop_transport_layers( connection_id, layer_count, @@ -2051,6 +2177,23 @@ fn redis_cluster_transport_id(connection_id: &str, endpoint: &db::redis_driver:: ) } +fn redis_sentinel_transport_prefix(connection_id: &str) -> String { + format!("{connection_id}:redis-sentinel:") +} + +fn redis_sentinel_transport_id( + connection_id: &str, + role: &str, + endpoint: &db::redis_driver::RedisNodeEndpoint, +) -> String { + format!( + "{}{role}:{host}:{port}", + redis_sentinel_transport_prefix(connection_id), + host = endpoint.host, + port = endpoint.port + ) +} + fn session_scoped_pool_key(base_pool_key: String, client_session_id: Option<&str>) -> String { normalize_client_session_id(client_session_id) .map(|session| format!("{base_pool_key}:session:{session}")) @@ -2430,8 +2573,9 @@ mod tests { use super::{ agent_connect_timeout, connection_remote_endpoint, connection_url_for_endpoint, database_connection_config, metadata_connection_config, mysql_metadata_fallback_url, oceanbase_mysql_setup_queries, - prestosql_jdbc_config_for_endpoint, redacted_connection_url_for_endpoint, uses_bare_mysql_pool, uses_tcp_probe, - validate_h2_database_path, AppState, PoolKind, PRESTOSQL_JDBC_DRIVER_CLASS, + prestosql_jdbc_config_for_endpoint, redacted_connection_url_for_endpoint, redis_sentinel_transport_id, + redis_sentinel_transport_prefix, uses_bare_mysql_pool, uses_tcp_probe, validate_h2_database_path, AppState, + PoolKind, PRESTOSQL_JDBC_DRIVER_CLASS, }; use crate::agent_connection::{ agent_connect_params, mongo_legacy_error_with_auth_hint, mongo_uses_legacy_driver, @@ -3226,6 +3370,17 @@ mod tests { ); } + #[test] + fn redis_sentinel_transport_ids_are_connection_scoped_by_role_and_endpoint() { + let endpoint = db::redis_driver::RedisNodeEndpoint { host: "10.0.0.8".to_string(), port: 6379 }; + + assert_eq!(redis_sentinel_transport_prefix("redis-prod"), "redis-prod:redis-sentinel:"); + assert_eq!( + redis_sentinel_transport_id("redis-prod", "master", &endpoint), + "redis-prod:redis-sentinel:master:10.0.0.8:6379" + ); + } + #[test] fn mysql_pool_size_keeps_session_pools_single_connection() { assert_eq!(super::mysql_pool_max_connections_for_session(None), 10); diff --git a/crates/dbx-core/src/db/redis_driver.rs b/crates/dbx-core/src/db/redis_driver.rs index 371d9438b2..5dd5237780 100644 --- a/crates/dbx-core/src/db/redis_driver.rs +++ b/crates/dbx-core/src/db/redis_driver.rs @@ -263,6 +263,38 @@ pub async fn connect_sentinel(config: &ConnectionConfig) -> Result Result { + let service_name = config.redis_sentinel_master.trim(); + if service_name.is_empty() { + return Err("Redis Sentinel master name is required".to_string()); + } + + let client = redis::Client::open(connection_info( + host, + port, + config.redis_sentinel_tls, + config.redis_tls_insecure(), + &config.redis_sentinel_username, + &config.redis_sentinel_password, + 0, + )) + .map_err(|e| format!("Redis Sentinel connect failed: {e}"))?; + let mut con = connect_client_with_timeout(client, super::connection_timeout(), "Redis Sentinel").await?; + let raw = tokio::time::timeout( + super::connection_timeout(), + redis::cmd("SENTINEL").arg("get-master-addr-by-name").arg(service_name).query_async::(&mut con), + ) + .await + .map_err(|_| format!("Redis Sentinel master lookup timed out ({}s)", super::CONNECTION_TIMEOUT_SECS))? + .map_err(|e| format!("Redis Sentinel master lookup failed: {e}"))?; + + redis_sentinel_master_endpoint(raw) +} + pub async fn connect_cluster(config: &ConnectionConfig) -> Result { let seed_nodes = redis_cluster_seed_nodes(config)?; let mut last_error = None; @@ -429,23 +461,30 @@ where } fn redis_sentinel_nodes(config: &ConnectionConfig) -> Result, String> { - let raw_nodes = config.redis_sentinel_nodes.trim(); - let endpoints: Vec = if raw_nodes.is_empty() { - vec![format!("{}:{}", config.host.trim(), config.port)] - } else { - raw_nodes - .split([',', ';', '\n', '\r']) - .map(str::trim) - .filter(|node| !node.is_empty()) - .map(ToOwned::to_owned) - .collect() - }; + redis_sentinel_node_endpoints(config)? + .iter() + .map(|endpoint| { + Ok(connection_info( + &endpoint.host, + endpoint.port, + config.redis_sentinel_tls, + config.redis_tls_insecure(), + &config.redis_sentinel_username, + &config.redis_sentinel_password, + 0, + )) + }) + .collect() +} - if endpoints.is_empty() { - return Err("At least one Redis Sentinel node is required".to_string()); - } - - endpoints.iter().map(|endpoint| redis_sentinel_node_connection_info(config, endpoint)).collect() +pub fn redis_sentinel_node_endpoints(config: &ConnectionConfig) -> Result, String> { + redis_node_endpoints( + config.redis_sentinel_nodes.trim(), + config.host.trim(), + config.port, + "Redis Sentinel node", + 26379, + ) } pub fn redis_cluster_seed_nodes(config: &ConnectionConfig) -> Result, String> { @@ -458,19 +497,6 @@ pub fn redis_cluster_seed_nodes(config: &ConnectionConfig) -> Result Result { - let (host, port) = parse_redis_endpoint(endpoint, 26379)?; - Ok(connection_info( - &host, - port, - config.redis_sentinel_tls, - config.redis_tls_insecure(), - &config.redis_sentinel_username, - &config.redis_sentinel_password, - 0, - )) -} - fn redis_node_endpoints( raw_nodes: &str, fallback_host: &str, @@ -502,6 +528,23 @@ fn redis_node_endpoints( .collect() } +fn redis_sentinel_master_endpoint(value: RedisRawValue) -> Result { + let RedisRawValue::Array(parts) = value else { + return Err("Redis Sentinel master lookup returned an invalid response".to_string()); + }; + if parts.len() != 2 { + return Err("Redis Sentinel master lookup returned an invalid endpoint".to_string()); + } + let Some(host) = redis_value_to_string(parts[0].clone()) else { + return Err("Redis Sentinel master lookup returned an invalid host".to_string()); + }; + let Some(port_text) = redis_value_to_string(parts[1].clone()) else { + return Err("Redis Sentinel master lookup returned an invalid port".to_string()); + }; + let port = parse_redis_port(&port_text)?; + Ok(RedisNodeEndpoint { host, port }) +} + fn connection_info( host: &str, port: u16, @@ -2329,8 +2372,8 @@ mod tests { parse_stream_entries, redis_auth_candidates, redis_cluster_slot, redis_command_raw_to_json, redis_database_index, redis_json_raw_to_json, redis_json_value_preview, redis_key_bytes_to_display, redis_key_bytes_to_raw, redis_key_matches_query, redis_key_raw_to_bytes, redis_key_value_preview, - redis_raw_to_json, redis_value_contains_binary, redis_value_matches_query, RedisAuthCandidate, - RedisClusterSlotRange, RedisCommandSafety, RedisNodeEndpoint, RedisRawValue, + redis_raw_to_json, redis_sentinel_master_endpoint, redis_value_contains_binary, redis_value_matches_query, + RedisAuthCandidate, RedisClusterSlotRange, RedisCommandSafety, RedisNodeEndpoint, RedisRawValue, }; use crate::models::connection::ConnectionConfig; use redis::ConnectionAddr; @@ -2579,6 +2622,26 @@ mod tests { assert_eq!(parse_redis_endpoint("::1", 26379).unwrap(), ("::1".to_string(), 26379)); } + #[test] + fn parses_redis_sentinel_master_lookup_response() { + let raw = RedisRawValue::Array(vec![bulk("10.0.0.8"), bulk("6379")]); + + assert_eq!( + redis_sentinel_master_endpoint(raw).unwrap(), + RedisNodeEndpoint { host: "10.0.0.8".to_string(), port: 6379 } + ); + } + + #[test] + fn rejects_invalid_redis_sentinel_master_lookup_response() { + let raw = RedisRawValue::Array(vec![bulk("10.0.0.8")]); + + assert_eq!( + redis_sentinel_master_endpoint(raw).unwrap_err(), + "Redis Sentinel master lookup returned an invalid endpoint" + ); + } + #[test] fn redis_connection_info_marks_tls_as_insecure_when_requested() { let info = connection_info("cache.example.com", 6379, true, true, "default", "secret", 0); diff --git a/src-tauri/src/commands/connection.rs b/src-tauri/src/commands/connection.rs index fc75bd1665..35dd7a62c3 100644 --- a/src-tauri/src/commands/connection.rs +++ b/src-tauri/src/commands/connection.rs @@ -559,7 +559,8 @@ pub async fn test_connection(state: State<'_, Arc>, config: Connection state.connect_redis_cluster(&tunnel_id, &config).await?; return Ok("Connection successful".to_string()); } else if config.uses_redis_sentinel() { - db::redis_driver::connect_sentinel(&config).await? + state.connect_redis_sentinel(&tunnel_id, &config).await?; + return Ok("Connection successful".to_string()); } else { db::redis_driver::connect(&url, connect_timeout).await? }; @@ -835,7 +836,7 @@ pub async fn connect_db(state: State<'_, Arc>, config: ConnectionConfi )) } else if db_config.uses_redis_sentinel() { PoolKind::Redis(db::redis_driver::RedisConnection::Direct(tokio::sync::Mutex::new( - db::redis_driver::connect_sentinel(&db_config).await?, + state.connect_redis_sentinel(&id, &db_config).await?, ))) } else { PoolKind::Redis(db::redis_driver::RedisConnection::Direct(tokio::sync::Mutex::new(