Fix tests again.

Author: ioquatix

cluster/test/async/redis/context/subscribe.rb

@@ -3,6 +3,7 @@
 # Released under the MIT License.
 # Copyright, 2025, by Samuel Williams.
 
+require "async/variable"
 require "async/redis/cluster_client"
 require "sus/fixtures/async"
 require "securerandom"
@@ -29,58 +30,34 @@
 
 			it "can subscribe to sharded channels and receive messages" do
 				received_message = nil
-				condition = Async::Condition.new
-				spublish_available = false
+				ready = Async::Variable.new
 
 				# Set up the subscriber using cluster client's ssubscribe method
-				subscriber_task = reactor.async do
-					begin
-						cluster.ssubscribe(shard_channel) do |context|
-							condition.signal # Signal that we're ready
-							spublish_available = true
-							
-							type, name, message = context.listen
-							
-							expect(type).to be == "smessage"
-							expect(name).to be == shard_channel
-							received_message = message
-						end
-					rescue Protocol::Redis::ServerError => error
-						if error.message.include?("unknown command")
-							Console.warn("SSUBSCRIBE not available on this Redis version")
-							condition.signal
-						else
-							raise
-						end
+				subscriber_task = Async do
+					cluster.ssubscribe(shard_channel) do |context|
+						ready.resolve
+						
+						type, name, message = context.listen
+						
+						expect(type).to be == "smessage"
+						expect(name).to be == shard_channel
+						received_message = message
 					end
 				end
 
 				# Set up the publisher
-				publisher_task = reactor.async do
-					condition.wait # Wait for subscriber to be ready
+				publisher_task = Async do
+					ready.wait
 
-					if spublish_available
-						begin
-							# Get a client on the same node as the subscriber for SPUBLISH
-							slot = cluster.slot_for(shard_channel)
-							publisher_client = cluster.client_for(slot)
-							publisher_client.call("SPUBLISH", shard_channel, shard_message)
-						rescue => error
-							Console.warn("SPUBLISH failed: #{error}")
-						end
-					end
+					slot = cluster.slot_for(shard_channel)
+					publisher_client = cluster.client_for(slot)
+					publisher_client.call("SPUBLISH", shard_channel, shard_message)
 				end
 
 				publisher_task.wait
-				sleep(0.1) # Allow message delivery
-				subscriber_task.stop
+				subscriber_task.wait
 
-				# Only check message if sharded pub/sub was available
-				if spublish_available && received_message
-					expect(received_message).to be == shard_message
-				else
-					Console.warn("Skipping assertion - sharded pub/sub not available on this Redis version")
-				end
+				expect(received_message).to be == shard_message
 			end
 
 			it "distributes sharded messages across cluster nodes" do
@@ -94,26 +71,26 @@
 				]
 
 				# Find channels that map to different slots/nodes
-				channel_slots = channels.map {|ch| [ch, cluster.slot_for(ch)]}
+				channel_slots = channels.map {|channel| [channel, cluster.slot_for(channel)]}
 				unique_slots = channel_slots.map(&:last).uniq
 
 				# We should have channels distributed across different slots
 				expect(unique_slots.size).to be > 1
 
 				received_messages = []
-				condition = Async::Condition.new
+				ready = Async::Variable.new
 				subscriber_count = 0
 				target_count = channels.size
 
 				# Set up subscribers for each channel
 				subscriber_tasks = channels.map do |channel|
-					reactor.async do
+					Async do
 						slot = cluster.slot_for(channel)
 						client = cluster.client_for(slot)
 
 						client.ssubscribe(channel) do |context|
 							subscriber_count += 1
-							condition.signal if subscriber_count == target_count
+							ready.resolve if subscriber_count == target_count
 
 							type, name, message = context.listen
 							received_messages << {channel: name, message: message, slot: slot}
@@ -122,27 +99,19 @@
 				end
 
 				# Set up publisher
-				publisher_task = reactor.async do
-					condition.wait # Wait for all subscribers
+				publisher_task = Async do
+					ready.wait # Wait for all subscribers
 
 					channels.each_with_index do |channel, index|
 						slot = cluster.slot_for(channel)
 						client = cluster.client_for(slot)
 
-						begin
-							client.call("SPUBLISH", channel, "message-#{index}")
-						rescue => error
-							Console.warn("SPUBLISH failed for #{channel}: #{error}")
-							# Clean up and skip if SPUBLISH not available
-							subscriber_tasks.each(&:stop)
-							return
-						end
+						client.call("SPUBLISH", channel, "message-#{index}")
 					end
 				end
 
 				publisher_task.wait
-				sleep(0.1) # Allow time for message delivery
-				subscriber_tasks.each(&:stop)
+				subscriber_tasks.each(&:wait)
 
 				# Verify we received messages for different channels
 				expect(received_messages.size).to be == channels.size
@@ -197,20 +166,12 @@
 
 					# Publish to sharded channel
 					shard_client = cluster.client_for(shard_slot)
-					begin
-						shard_client.call("SPUBLISH", shard_channel, "sharded message")
-					rescue => error
-						Console.warn("SPUBLISH not available: #{error}")
-						regular_task.stop
-						shard_task.stop
-						return
-					end
+					shard_client.call("SPUBLISH", shard_channel, "sharded message")
 				end
 
 				publisher_task.wait
-				sleep(0.1) # Allow time for message delivery
-				regular_task.stop
-				shard_task.stop
+				regular_task.wait
+				shard_task.wait
 
 				# Should have received both messages
 				expect(received_messages.size).to be == 2
@@ -276,17 +237,11 @@
 					publisher_client.publish(channel, "unified regular")
 
 					# Publish sharded message
-					begin
-						publisher_client.call("SPUBLISH", shard_channel, "unified sharded")
-					rescue => error
-						Console.warn("SPUBLISH not available: #{error}")
-						# Skip sharded part if not available
-					end
+					publisher_client.call("SPUBLISH", shard_channel, "unified sharded")
 				end
 
 				publisher_task.wait
-				sleep(0.1) # Allow message delivery
-				subscriber_task.stop
+				subscriber_task.wait
 
 				# Should receive both message types on same context
 				expect(received_messages.size).to be == 2

lib/async/redis/cluster_client.rb

@@ -63,6 +63,16 @@ def each
 					end
 				end
 
+				# Get a random value from the map.
+				# @returns [Object] A randomly selected value, or nil if map is empty.
+				def sample
+					return nil if @ranges.empty?
+					
+					range, value = @ranges.sample
+					
+					return value
+				end
+				
 				# Clear all ranges from the map.
 				def clear
 					@ranges.clear
@@ -127,6 +137,38 @@ def client_for(slot, role = :master)
 				end
 			end
 
+			# Get any available client from the cluster.
+			# This is useful for operations that don't require slot-specific routing,
+			# such as global pub/sub operations, INFO commands, or other cluster-wide operations.
+			# @parameter role [Symbol] The role of node to get (:master or :slave).
+			# @returns [Client] A Redis client for any available node.
+			def any_client(role = :master)
+				unless @shards
+					reload_cluster!
+				end
+				
+				# Sample a random shard to get better load distribution
+				if nodes = @shards.sample
+					nodes = nodes.select{|node| node.role == role}
+					
+					if node = nodes.sample
+						return (node.client ||= Client.new(node.endpoint, **@options))
+					end
+				end
+				
+				# Fallback to slot 0 if sampling fails
+				client_for(0, role)
+			end
+			
+			# Execute a Redis command on any available cluster node.
+			# This is useful for commands that don't require slot-specific routing.
+			# @parameter command [String] The Redis command to execute.
+			# @parameter arguments [Array] The command arguments.
+			# @returns [Object] The result of the Redis command.
+			def call(command, *arguments)
+				any_client.call(command, *arguments)
+			end
+			
 			protected
 
 			def reload_cluster!(endpoints = @endpoints)
@@ -245,16 +287,22 @@ def slots_for(keys)
 			end
 
 			# Subscribe to one or more channels for pub/sub messaging in cluster environment.
-			# The subscription will be created on the node responsible for the first channel.
+			# 
+			# NOTE: Regular pub/sub in Redis Cluster is GLOBAL - messages propagate to all nodes.
+			# This method is a convenience that subscribes via an arbitrary cluster node.
+			# The choice of node does not affect which messages you receive, since regular
+			# pub/sub messages are broadcast to all nodes in the cluster.
+			# 
+			# For slot-aware pub/sub, use ssubscribe() instead (Redis 7.0+).
+			# 
 			# @parameter channels [Array(String)] The channels to subscribe to.
 			# @yields {|context| ...} If a block is given, it will be executed within the subscription context.
 			# 	@parameter context [Context::Subscribe] The subscription context.
 			# @returns [Object] The result of the block if block given.
 			# @returns [Context::Subscribe] The subscription context if no block given.
 			def subscribe(*channels)
-				# For regular pub/sub, route to node based on first channel
-				slot = channels.any? ? slot_for(channels.first) : 0
-				client = client_for(slot)
+				# For regular pub/sub, use any available node since messages are global
+				client = any_client
 
 				client.subscribe(*channels) do |context|
 					if block_given?
@@ -266,16 +314,20 @@ def subscribe(*channels)
 			end
 
 			# Subscribe to one or more channel patterns for pub/sub messaging in cluster environment.
-			# The subscription will be created on the node responsible for a deterministic slot.
+			# 
+			# NOTE: Pattern subscriptions in Redis Cluster are GLOBAL - they match channels
+			# across all nodes. This method is a convenience that subscribes via an arbitrary
+			# cluster node. Pattern matching works across the entire cluster regardless of
+			# which node you subscribe from.
+			# 
 			# @parameter patterns [Array(String)] The channel patterns to subscribe to.
 			# @yields {|context| ...} If a block is given, it will be executed within the subscription context.
 			# 	@parameter context [Context::Subscribe] The subscription context.
 			# @returns [Object] The result of the block if block given.
 			# @returns [Context::Subscribe] The subscription context if no block given.
 			def psubscribe(*patterns)
-				# For pattern subscriptions, use a deterministic slot since patterns can match any channel
-				slot = patterns.any? ? slot_for(patterns.first) : 0
-				client = client_for(slot)
+				# For pattern subscriptions, use any available node since patterns are global
+				client = any_client
 
 				client.psubscribe(*patterns) do |context|
 					if block_given?