Merge branch 'main' into task/add-scripting-commands

Author: prashanna-frsh

CHANGELOG.md

@@ -11,11 +11,15 @@
 * CORE: Add client certificate and private key support for mTLS ([#5092](https://github.com/valkey-io/valkey-glide/issues/5092))
 * Python: Add client certificate and private key support for mTLS ([5123](https://github.com/valkey-io/valkey-glide/issues/5123))
 * JAVA: Add KEYS, MIGRATE and WAITAOF command support ([#5017](https://github.com/valkey-io/valkey-glide/pull/5107))
+* Python Sync: Add dynamic PubSub methods (subscribe, psubscribe, unsubscribe, punsubscribe, ssubscribe, sunsubscribe with timeout support), get_subscriptions(), and pubsub_reconciliation_interval configuration support. Unified configuration classes with async client by importing from glide_shared ([#5270](https://github.com/valkey-io/valkey-glide/pull/5270))
+* Go: Add dynamic PubSub methods (Subscribe, PSubscribe, SSubscribe, Unsubscribe, PUnsubscribe, SUnsubscribe with blocking variants and timeout support), GetSubscriptions(), and helper functions (UnsubscribeAll, PUnsubscribeAll, SUnsubscribeAll). Export pubsub constants (AllChannels, AllPatterns, AllShardedChannels)
 * JAVA: Add ACL server management commands (ACL CAT, ACL DELUSER, ACL DRYRUN, ACL GENPASS, ACL GETUSER, ACL LIST, ACL LOAD, ACL LOG, ACL SAVE, ACL SETUSER, ACL USERS, ACL WHOAMI)
 * CORE: Fix typo in ACL SETUSER command mapping (AclSetSser → AclSetUser)
 * Python: Add inflight request limit support to sync client
 * Python Sync: Add OpenTelemetry support with traces and metrics configuration
 * Python: Move OpenTelemetry config classes to glide_shared for code reuse between async and sync clients
+* JAVA: Add dynamic PubSub methods (subscribe, psubscribe, unsubscribe, punsubscribe, ssubscribe, sunsubscribe), getSubscriptions() for subscription state tracking, pubsubReconciliationIntervalMs configuration option, and subscription_out_of_sync_count and subscription_last_sync_timestamp metrics ([#5267](https://github.com/valkey-io/valkey-glide/issues/5267))
+* Go: Add ALLOW_NON_COVERED_SLOTS flag support for cluster scan ([#4895](https://github.com/valkey-io/valkey-glide/issues/4895))
 
 #### Fixes
 * Node: Fix to handle non-string types in toBuffersArray ([#4842](https://github.com/valkey-io/valkey-glide/issues/4842))

ffi/miri-tests/mock-telemetry/src/lib.rs

@@ -124,6 +124,10 @@ impl Telemetry {
     pub fn total_bytes_decompressed() -> usize { 0 }
     pub fn incr_compression_skipped_count(_incr_by: usize) -> usize { 0 }
     pub fn compression_skipped_count() -> usize { 0 }
+    pub fn incr_subscription_out_of_sync() -> usize { 0 }
+    pub fn subscription_out_of_sync_count() -> usize { 0 }
+    pub fn update_subscription_last_sync_timestamp(_timestamp: u64) -> u64 { 0 }
+    pub fn subscription_last_sync_timestamp() -> u64 { 0 }
     pub fn reset() {}
 }
 

ffi/src/lib.rs

@@ -3004,6 +3004,10 @@ pub struct Statistics {
     pub total_bytes_decompressed: c_ulong,
     /// Number of times compression was skipped
     pub compression_skipped_count: c_ulong,
+    /// Number of times subscriptions were out of sync during reconciliation
+    pub subscription_out_of_sync_count: c_ulong,
+    /// Timestamp of last successful subscription sync (milliseconds since epoch)
+    pub subscription_last_sync_timestamp: c_ulong,
 }
 
 /// Get compression and connection statistics.
@@ -3025,6 +3029,8 @@ pub extern "C" fn get_statistics() -> Statistics {
         total_bytes_compressed: Telemetry::total_bytes_compressed() as c_ulong,
         total_bytes_decompressed: Telemetry::total_bytes_decompressed() as c_ulong,
         compression_skipped_count: Telemetry::compression_skipped_count() as c_ulong,
+        subscription_out_of_sync_count: Telemetry::subscription_out_of_sync_count() as c_ulong,
+        subscription_last_sync_timestamp: Telemetry::subscription_last_sync_timestamp() as c_ulong,
     }
 }
 

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

@@ -2213,79 +2213,82 @@ where
         let nodes = new_slots.all_node_addresses();
         let nodes_len = nodes.len();
 
-        // Find existing connections or resolve DNS addresses
-        // TODO: optimize by parallelizing DNS lookups - DNS resolution may involve network I/O and is currently performed sequentially
-        // Issue link: https://github.com/valkey-io/valkey-glide/issues/5228
-        let mut addresses_and_connections = Vec::with_capacity(nodes_len);
-        for addr in nodes {
-            let addr = addr.to_string();
-
-            // Check for existing connection by address
-            if let Some(node) = inner
-                .conn_lock
-                .read()
-                .expect(MUTEX_READ_ERR)
-                .node_for_address(&addr)
-            {
-                addresses_and_connections.push((addr, Some(node)));
-                continue;
-            }
-
-            // If it's a DNS endpoint, it could have been stored in the existing connections vector using the resolved IP address instead of the DNS endpoint's name.
-            // We shall check if a connection is already exists under the resolved IP name.
-            let conn = if let Some((host, port)) = get_host_and_port_from_addr(&addr) {
-                get_socket_addrs(host, port)
-                    .await
-                    .ok()
-                    .and_then(|mut socket_addresses| {
-                        let conn_lock = inner.conn_lock.read().expect(MUTEX_READ_ERR);
-                        socket_addresses.find_map(|socket_addr| {
-                            conn_lock.node_for_address(&socket_addr.to_string())
-                        })
-                    })
-            } else {
-                None
-            };
-
-            // If we found a connection by IP lookup, update the PushManager. This ensures the PushManager
-            // stores the DNS address (which matches the connection_map key) instead of the old IP
-            // or config endpoint address, which is needed for pubsub tracking
-            if let Some(ref node) = conn {
-                node.user_connection
-                    .conn
-                    .clone()
-                    .await
-                    .update_push_manager_node_address(addr.clone());
-            }
-
-            addresses_and_connections.push((addr, conn));
-        }
-
         let cluster_params = inner
             .get_cluster_param(|params| params.clone())
             .expect(MUTEX_READ_ERR);
         let glide_connection_options = &inner.glide_connection_options;
 
-        // Create/retrieve connections in for found nodes
-        let connection_futures = addresses_and_connections.into_iter().map(|(addr, node)| {
+        // Find existing connections (by address or DNS resolution) or create new ones
+        let connection_futures = nodes.into_iter().map(|addr| {
+            let addr = addr.to_string();
+            let inner = Arc::clone(&inner);
             let cluster_params = cluster_params.clone();
             let glide_connection_options = glide_connection_options.clone();
+            let connection_timeout = cluster_params.connection_timeout;
+
             async move {
-                let result = get_or_create_conn(
-                    &addr,
-                    node,
-                    &cluster_params,
-                    RefreshConnectionType::AllConnections,
-                    glide_connection_options,
-                )
-                .await;
+                // TODO: Expose separate `dns_timeout` configuration in advanced settings
+                // to allow users to control DNS resolution timeout independently from connection timeout.
+                // Issue: https://github.com/valkey-io/valkey-glide/issues/5298
+                let result = tokio::time::timeout(connection_timeout, async {
+                    // Check for existing connection by direct address
+                    let node = inner
+                        .conn_lock
+                        .read()
+                        .expect(MUTEX_READ_ERR)
+                        .node_for_address(&addr);
+
+                    let node = match node {
+                        Some(n) => Some(n),
+                        None => {
+                            // If it's a DNS endpoint, it could have been stored in the existing connections vector
+                            // using the resolved IP address instead of the DNS endpoint's name.
+                            // We shall check if a connection already exists under the resolved IP name.
+                            if let Some((host, port)) = get_host_and_port_from_addr(&addr) {
+                                let conn = get_socket_addrs(host, port).await.ok().and_then(
+                                    |mut socket_addresses| {
+                                        let conn_lock =
+                                            inner.conn_lock.read().expect(MUTEX_READ_ERR);
+                                        socket_addresses.find_map(|socket_addr| {
+                                            conn_lock.node_for_address(&socket_addr.to_string())
+                                        })
+                                    },
+                                );
+
+                                // If we found a connection by IP lookup, update the PushManager. This ensures
+                                // the PushManager stores the DNS address (which matches the connection_map key)
+                                // instead of the old IP or config endpoint address, which is needed for pubsub tracking.
+                                if let Some(ref node) = conn {
+                                    node.user_connection
+                                        .conn
+                                        .clone()
+                                        .await
+                                        .update_push_manager_node_address(addr.clone());
+                                }
+                                conn
+                            } else {
+                                None
+                            }
+                        }
+                    };
+
+                    get_or_create_conn(
+                        &addr,
+                        node,
+                        &cluster_params,
+                        RefreshConnectionType::AllConnections,
+                        glide_connection_options,
+                    )
+                    .await
+                })
+                .await
+                .unwrap_or_else(|e| Err(e.into()));
+
                 (addr, result)
             }
         });
 
-        // Await all connection futures
-        // This is bounded by `connection_timeout` since `get_or_create_conn` uses it internally,
-        // so it is safe to await them all at once.
+        // Await all connection futures, this is bounded by `connection_timeout`.
         let results = futures::future::join_all(connection_futures).await;
 
         // Collect successful connections

go/README.md

@@ -139,8 +139,154 @@ func main() {
 }
 ```
 
+### PubSub Example:
+
+Valkey GLIDE supports PubSub (Publish/Subscribe) for real-time messaging. You can subscribe to channels and patterns, and dynamically manage subscriptions.
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "time"
+
+    glide "github.com/valkey-io/valkey-glide/go/v2"
+    "github.com/valkey-io/valkey-glide/go/v2/config"
+)
+
+func main() {
+    host := "localhost"
+    port := 6379
+
+    // Configure client with PubSub callback and reconciliation interval
+    config := config.NewClientConfiguration().
+        WithAddress(&config.NodeAddress{Host: host, Port: port}).
+        WithPubSubCallback(func(msg glide.PubSubMessage, ctx glide.MessageContext) {
+            fmt.Printf("Received message: %s from channel: %s\n", msg.Message, msg.Channel)
+        }).
+        WithPubSubReconciliationInterval(5000) // Reconcile subscriptions every 5 seconds
+
+    client, err := glide.NewClient(config)
+    if err != nil {
+        fmt.Println("Error creating client:", err)
+        return
+    }
+    defer client.Close()
+
+    ctx := context.Background()
+
+    // Subscribe to channels
+    err = client.Subscribe(ctx, []string{"news", "updates"})
+    if err != nil {
+        fmt.Println("Error subscribing:", err)
+        return
+    }
+
+    // Subscribe to patterns
+    err = client.PSubscribe(ctx, []string{"events:*"})
+    if err != nil {
+        fmt.Println("Error pattern subscribing:", err)
+        return
+    }
+
+    // Get current subscriptions
+    state, err := client.GetSubscriptions(ctx)
+    if err != nil {
+        fmt.Println("Error getting subscriptions:", err)
+        return
+    }
+    fmt.Printf("Subscribed to %d channels and %d patterns\n", 
+        len(state.DesiredSubscriptions["exact"]), 
+        len(state.DesiredSubscriptions["pattern"]))
+
+    // Wait for messages
+    time.Sleep(10 * time.Second)
+
+    // Unsubscribe from specific channels
+    err = client.Unsubscribe(ctx, []string{"news"})
+    if err != nil {
+        fmt.Println("Error unsubscribing:", err)
+        return
+    }
+
+    // Unsubscribe from all channels and patterns
+    err = client.UnsubscribeAll(ctx)
+    if err != nil {
+        fmt.Println("Error unsubscribing from all:", err)
+        return
+    }
+}
+```
+
+**PubSub Configuration Options:**
+
+- `WithPubSubCallback`: Set a callback function to handle incoming PubSub messages
+- `WithPubSubReconciliationInterval`: Set the interval (in milliseconds) for automatic subscription reconciliation. This ensures subscriptions are maintained even if the connection is temporarily lost. Default is 0 (disabled).
+
+**PubSub Methods:**
+
+- `Subscribe(ctx, channels)`: Subscribe to one or more channels
+- `PSubscribe(ctx, patterns)`: Subscribe to channel patterns (e.g., "news:*")
+- `SSubscribe(ctx, channels)`: Subscribe to sharded channels (cluster mode only)
+- `Unsubscribe(ctx, channels)`: Unsubscribe from specific channels (pass `nil` or `glide.AllChannels` to unsubscribe from all)
+- `PUnsubscribe(ctx, patterns)`: Unsubscribe from patterns (pass `nil` or `glide.AllPatterns` to unsubscribe from all)
+- `SUnsubscribe(ctx, channels)`: Unsubscribe from sharded channels (pass `nil` or `glide.AllShardedChannels` to unsubscribe from all)
+- `GetSubscriptions(ctx)`: Get the current subscription state
+
+**Blocking Variants:**
+
+All subscribe/unsubscribe methods have blocking variants with timeout support:
+- `SubscribeBlocking(ctx, channels, timeoutMs)`
+- `UnsubscribeBlocking(ctx, channels, timeoutMs)`
+- And similar for PSubscribe, SSubscribe, etc.
+
 For more code examples please refer to [examples.md](examples/examples.md).
 
+### Cluster Scan
+
+The cluster scan feature allows you to iterate over all keys in a cluster. You can optionally filter by pattern, type, and control batch size.
+
+#### Basic Cluster Scan
+
+```go
+cursor := models.NewClusterScanCursor()
+allKeys := []string{}
+
+for !cursor.IsFinished() {
+    result, err := client.Scan(context.Background(), cursor)
+    if err != nil {
+        fmt.Println("Error:", err)
+        break
+    }
+    allKeys = append(allKeys, result.Keys...)
+    cursor = result.Cursor
+}
+```
+
+#### Cluster Scan with Options
+
+```go
+opts := options.NewClusterScanOptions().
+    SetMatch("user:*").              // Filter by pattern
+    SetCount(100).                   // Batch size hint
+    SetType(constants.StringType).   // Filter by key type
+    SetAllowNonCoveredSlots(true)    // Allow scanning even if some slots are not covered
+
+cursor := models.NewClusterScanCursor()
+for !cursor.IsFinished() {
+    result, err := client.ScanWithOptions(context.Background(), cursor, *opts)
+    if err != nil {
+        fmt.Println("Error:", err)
+        break
+    }
+    // Process result.Keys
+    cursor = result.Cursor
+}
+```
+
+**Note**: The `AllowNonCoveredSlots` option is useful when the cluster is not fully configured or some nodes are down. It allows the scan to proceed even if some hash slots are not covered by any node.
+
 ### Building & Testing
 
 Development instructions for local building & testing the package are in the [DEVELOPER.md](DEVELOPER.md) file.