Merge remote-tracking branch 'origin/fix-test_pubsub_combined_exact_pattern_and_sharded_one_client' into fix-pubsub-deterministic-channels

Merge valkey-io#4886 into valkey-io#4880 to combine flaky Pub/Sub test fixes

Includes fixes from fix-test_pubsub_combined_exact_pattern_and_sharded_one_client
into fix-pubsub-deterministic-channels as requested in issue valkey-io#4862.
Stabilizes all related Pub/Sub tests with unique channel names and proper sync.#

Author: jh1231223

CHANGELOG.md

@@ -3,6 +3,9 @@
 #### Changes
 
 * Added in documentation to how to use Lua scripts with Glide 
+* JAVA: Add refreshTopologyFromInitialNodes Configuration Option ([#4870](https://github.com/valkey-io/valkey-glide/pull/4870))
+* GO: add RefreshTopologyFromInitialNodes configuration option ([#4871](https://github.com/valkey-io/valkey-glide/pull/4871))
+* Node: add refreshTopologyFromInitialNodes configuration option ([#4872](https://github.com/valkey-io/valkey-glide/pull/4872))
 
 #### Fixes
 

glide-core/redis-rs/redis/src/aio/mod.rs

@@ -227,9 +227,10 @@ where
 
     // result is ignored, as per the command's instructions.
     // https://redis.io/commands/client-setinfo/
-    let _: RedisResult<()> = crate::connection::client_set_info_pipeline()
-        .query_async(con)
-        .await;
+    let _: RedisResult<()> =
+        crate::connection::client_set_info_pipeline(connection_info.lib_name.as_deref())
+            .query_async(con)
+            .await;
 
     // resubscribe
     if connection_info.protocol != ProtocolVersion::RESP3 {

glide-core/redis-rs/redis/src/cluster.rs

@@ -991,6 +991,7 @@ pub(crate) fn get_connection_info(
             password: cluster_params.password,
             username: cluster_params.username,
             client_name: cluster_params.client_name,
+            lib_name: cluster_params.lib_name,
             protocol: cluster_params.protocol,
             db: cluster_params.database_id,
             pubsub_subscriptions: cluster_params.pubsub_subscriptions,

glide-core/redis-rs/redis/src/cluster_client.rs

@@ -40,6 +40,7 @@ struct BuilderParams {
     #[cfg(feature = "cluster-async")]
     slots_refresh_rate_limit: SlotsRefreshRateLimit,
     client_name: Option<String>,
+    lib_name: Option<String>,
     response_timeout: Option<Duration>,
     protocol: ProtocolVersion,
     pubsub_subscriptions: Option<PubSubSubscriptionInfo>,
@@ -141,6 +142,7 @@ pub struct ClusterParams {
     pub(crate) connections_validation_interval: Option<Duration>,
     pub(crate) tls_params: Option<TlsConnParams>,
     pub(crate) client_name: Option<String>,
+    pub(crate) lib_name: Option<String>,
     pub(crate) connection_timeout: Duration,
     pub(crate) response_timeout: Duration,
     pub(crate) protocol: ProtocolVersion,
@@ -173,6 +175,7 @@ impl ClusterParams {
             connections_validation_interval: value.connections_validation_interval,
             tls_params,
             client_name: value.client_name,
+            lib_name: value.lib_name,
             response_timeout: value.response_timeout.unwrap_or(Duration::MAX),
             protocol: value.protocol,
             pubsub_subscriptions: value.pubsub_subscriptions,
@@ -308,6 +311,12 @@ impl ClusterClientBuilder {
         self
     }
 
+    /// Sets library name for the new ClusterClient.
+    pub fn lib_name(mut self, lib_name: String) -> ClusterClientBuilder {
+        self.builder_params.lib_name = Some(lib_name);
+        self
+    }
+
     /// Sets password for the new ClusterClient.
     pub fn password(mut self, password: String) -> ClusterClientBuilder {
         self.builder_params.password = Some(password);

glide-core/redis-rs/redis/src/connection.rs

@@ -232,6 +232,8 @@ pub struct RedisConnectionInfo {
     pub protocol: ProtocolVersion,
     /// Optionally a client name that should be used for connection
     pub client_name: Option<String>,
+    /// Optionally a library name that should be used for connection
+    pub lib_name: Option<String>,
     /// Optionally a pubsub subscriptions that should be used for connection
     pub pubsub_subscriptions: Option<PubSubSubscriptionInfo>,
 }
@@ -391,6 +393,7 @@ fn url_to_tcp_connection_info(url: url::Url) -> RedisResult<ConnectionInfo> {
                 _ => ProtocolVersion::RESP2,
             },
             client_name: None,
+            lib_name: None,
             pubsub_subscriptions: None,
         },
     })
@@ -424,6 +427,7 @@ fn url_to_unix_connection_info(url: url::Url) -> RedisResult<ConnectionInfo> {
                 _ => ProtocolVersion::RESP2,
             },
             client_name: None,
+            lib_name: None,
             pubsub_subscriptions: None,
         },
     })
@@ -925,13 +929,15 @@ pub fn connect(
     setup_connection(con, &connection_info.redis)
 }
 
-pub(crate) fn client_set_info_pipeline() -> Pipeline {
+pub(crate) fn client_set_info_pipeline(lib_name: Option<&str>) -> Pipeline {
     let mut pipeline = crate::pipe();
+    let lib_name_value = lib_name.unwrap_or("UnknownClient");
+    let final_lib_name = option_env!("GLIDE_NAME").unwrap_or(lib_name_value);
     pipeline
         .cmd("CLIENT")
         .arg("SETINFO")
         .arg("LIB-NAME")
-        .arg(std::env!("GLIDE_NAME"))
+        .arg(final_lib_name)
         .ignore();
     pipeline
         .cmd("CLIENT")
@@ -993,7 +999,8 @@ fn setup_connection(
 
     // result is ignored, as per the command's instructions.
     // https://redis.io/commands/client-setinfo/
-    let _: RedisResult<()> = client_set_info_pipeline().query(&mut rv);
+    let _: RedisResult<()> =
+        client_set_info_pipeline(connection_info.lib_name.as_deref()).query(&mut rv);
 
     Ok(rv)
 }
@@ -1726,6 +1733,34 @@ pub fn get_resp3_hello_command_error(err: RedisError) -> RedisError {
 mod tests {
     use super::*;
 
+    #[test]
+    fn test_client_set_info_pipeline_default_lib_name() {
+        let pipeline = client_set_info_pipeline(None);
+        let packed_commands = pipeline.get_packed_pipeline();
+        let cmd_str = String::from_utf8_lossy(&packed_commands);
+
+        // Should contain CLIENT SETINFO LIB-NAME
+        assert!(cmd_str.contains("CLIENT"));
+        assert!(cmd_str.contains("SETINFO"));
+        assert!(cmd_str.contains("LIB-NAME"));
+
+        // When GLIDE_NAME is set, it should use that value
+        // When GLIDE_NAME is not set and lib_name is None, it should use "UnknownClient"
+        // Since we can't control GLIDE_NAME in this test, we just verify the structure
+        assert!(cmd_str.contains("Glide") || cmd_str.contains("UnknownClient"));
+    }
+
+    #[test]
+    fn test_client_set_info_pipeline_logic() {
+        // Test the logic directly by simulating what happens when GLIDE_NAME is not set
+        let lib_name_value = None.unwrap_or("UnknownClient");
+        assert_eq!(lib_name_value, "UnknownClient");
+
+        // Test with provided lib_name
+        let lib_name_value = Some("CustomClient").unwrap_or("UnknownClient");
+        assert_eq!(lib_name_value, "CustomClient");
+    }
+
     #[test]
     fn test_parse_redis_url() {
         let cases = vec![
@@ -1840,6 +1875,7 @@ mod tests {
                         password: None,
                         protocol: ProtocolVersion::RESP2,
                         client_name: None,
+                        lib_name: None,
                         pubsub_subscriptions: None,
                     },
                 },

glide-core/src/client/mod.rs

@@ -145,6 +145,7 @@ pub async fn get_valkey_connection_info(
     let protocol = connection_request.protocol.unwrap_or_default();
     let db = connection_request.database_id;
     let client_name = connection_request.client_name.clone();
+    let lib_name = connection_request.lib_name.clone();
     let pubsub_subscriptions = connection_request.pubsub_subscriptions.clone();
 
     match &connection_request.authentication_info {
@@ -164,6 +165,7 @@ pub async fn get_valkey_connection_info(
                     password: Some(token),
                     protocol,
                     client_name,
+                    lib_name,
                     pubsub_subscriptions,
                 }
             } else {
@@ -174,6 +176,7 @@ pub async fn get_valkey_connection_info(
                     password: info.password.clone(),
                     protocol,
                     client_name,
+                    lib_name,
                     pubsub_subscriptions,
                 }
             }
@@ -182,6 +185,7 @@ pub async fn get_valkey_connection_info(
             db,
             protocol,
             client_name,
+            lib_name,
             pubsub_subscriptions,
             ..Default::default()
         },
@@ -1052,6 +1056,9 @@ async fn create_cluster_client(
     if let Some(client_name) = valkey_connection_info.client_name {
         builder = builder.client_name(client_name);
     }
+    if let Some(lib_name) = valkey_connection_info.lib_name {
+        builder = builder.lib_name(lib_name);
+    }
     if tls_mode != TlsMode::NoTls {
         let tls = if tls_mode == TlsMode::SecureTls {
             redis::cluster::TlsMode::Secure

glide-core/src/client/types.rs

@@ -14,6 +14,7 @@ use crate::iam::ServiceType;
 pub struct ConnectionRequest {
     pub read_from: Option<ReadFrom>,
     pub client_name: Option<String>,
+    pub lib_name: Option<String>,
     pub authentication_info: Option<AuthenticationInfo>,
     pub database_id: i64,
     pub protocol: Option<redis::ProtocolVersion>,
@@ -164,6 +165,7 @@ impl From<protobuf::ConnectionRequest> for ConnectionRequest {
         });
 
         let client_name = chars_to_string_option(&value.client_name);
+        let lib_name = chars_to_string_option(&value.lib_name);
         let authentication_info = value.authentication_info.0.map(|authentication_info| {
             let password = chars_to_string_option(&authentication_info.password);
             let username = chars_to_string_option(&authentication_info.username);
@@ -278,6 +280,7 @@ impl From<protobuf::ConnectionRequest> for ConnectionRequest {
         ConnectionRequest {
             read_from,
             client_name,
+            lib_name,
             authentication_info,
             database_id,
             protocol,

glide-core/src/protobuf/connection_request.proto

@@ -88,6 +88,7 @@ message ConnectionRequest {
     uint32 connection_timeout = 16;
     bool lazy_connect = 17;
     bool refresh_topology_from_initial_nodes = 18;
+    string lib_name = 19;
 }
 
 message ConnectionRetryStrategy {

go/config/config.go

@@ -397,6 +397,7 @@ func (config *ClusterClientConfiguration) ToProtobuf() (*protobuf.ConnectionRequ
 	if config.subscriptionConfig != nil && len(config.subscriptionConfig.subscriptions) > 0 {
 		request.PubsubSubscriptions = config.subscriptionConfig.toProtobuf()
 	}
+	request.RefreshTopologyFromInitialNodes = config.AdvancedClusterClientConfiguration.refreshTopologyFromInitialNodes
 	return request, nil
 }
 
@@ -543,7 +544,8 @@ func (config *AdvancedClientConfiguration) WithConnectionTimeout(
 // Represents advanced configuration settings for a Cluster client used in
 // [ClusterClientConfiguration].
 type AdvancedClusterClientConfiguration struct {
-	connectionTimeout time.Duration
+	connectionTimeout               time.Duration
+	refreshTopologyFromInitialNodes bool
 }
 
 // NewAdvancedClusterClientConfiguration returns a new [AdvancedClusterClientConfiguration] with default settings.
@@ -561,3 +563,17 @@ func (config *AdvancedClusterClientConfiguration) WithConnectionTimeout(
 	config.connectionTimeout = connectionTimeout
 	return config
 }
+
+// WithRefreshTopologyFromInitialNodes enables refreshing the cluster topology using only the initial nodes.
+//
+// When this option is enabled, all topology updates (both the periodic checks and on-demand
+// refreshes triggered by topology changes) will query only the initial nodes provided when
+// creating the client, rather than using the internal cluster view.
+//
+// If not set, defaults to false (uses internal cluster view for topology refresh).
+func (config *AdvancedClusterClientConfiguration) WithRefreshTopologyFromInitialNodes(
+	refreshTopologyFromInitialNodes bool,
+) *AdvancedClusterClientConfiguration {
+	config.refreshTopologyFromInitialNodes = refreshTopologyFromInitialNodes
+	return config
+}

go/config/config_test.go

@@ -391,3 +391,37 @@ func TestConfig_DatabaseId_BaseConfiguration(t *testing.T) {
 	assert.Equal(t, uint32(3), clusterResult.DatabaseId)
 	assert.True(t, clusterResult.ClusterModeEnabled)
 }
+
+func TestClusterConfig_RefreshTopologyFromInitialNodes(t *testing.T) {
+	// Test that refreshTopologyFromInitialNodes defaults to false
+	defaultConfig := NewClusterClientConfiguration()
+	defaultResult, err := defaultConfig.ToProtobuf()
+	if err != nil {
+		t.Fatalf("Failed to convert default cluster config to protobuf: %v", err)
+	}
+	assert.False(t, defaultResult.RefreshTopologyFromInitialNodes)
+
+	// Test that refreshTopologyFromInitialNodes can be set to true
+	enabledConfig := NewClusterClientConfiguration().
+		WithAdvancedConfiguration(
+			NewAdvancedClusterClientConfiguration().
+				WithRefreshTopologyFromInitialNodes(true),
+		)
+	enabledResult, err := enabledConfig.ToProtobuf()
+	if err != nil {
+		t.Fatalf("Failed to convert enabled cluster config to protobuf: %v", err)
+	}
+	assert.True(t, enabledResult.RefreshTopologyFromInitialNodes)
+
+	// Test that refreshTopologyFromInitialNodes can be explicitly set to false
+	disabledConfig := NewClusterClientConfiguration().
+		WithAdvancedConfiguration(
+			NewAdvancedClusterClientConfiguration().
+				WithRefreshTopologyFromInitialNodes(false),
+		)
+	disabledResult, err := disabledConfig.ToProtobuf()
+	if err != nil {
+		t.Fatalf("Failed to convert disabled cluster config to protobuf: %v", err)
+	}
+	assert.False(t, disabledResult.RefreshTopologyFromInitialNodes)
+}