One pool per node test

hedis.cabal

@@ -90,7 +90,8 @@ library
                     HTTP,
                     errors,
                     network-uri,
-                    unliftio-core
+                    unliftio-core,
+                    random
   if !impl(ghc >= 8.0)
     build-depends:
       semigroups >= 0.11 && < 0.19

src/Database/Redis/Cluster.hs

@@ -12,6 +12,7 @@ module Database.Redis.Cluster
   , ShardMap(..)
   , HashSlot
   , Shard(..)
+  , TimeoutException(..)
   , connect
   , disconnect
   , requestPipelined
@@ -23,7 +24,7 @@ import qualified Data.ByteString as B
 import qualified Data.ByteString.Char8 as Char8
 import qualified Data.IORef as IOR
 import Data.Maybe(mapMaybe, fromMaybe)
-import Data.List(nub, sortBy, find)
+import Data.List(nub, sortBy, find, isPrefixOf)
 import Data.Map(fromListWith, assocs)
 import Data.Function(on)
 import Control.Exception(Exception, SomeException, throwIO, BlockedIndefinitelyOnMVar(..), catches, Handler(..), try, displayException)
@@ -124,7 +125,7 @@ instance Exception CrossSlotException
 data NoNodeException = NoNodeException  deriving (Show, Typeable)
 instance Exception NoNodeException
 
-data TimeoutException = TimeoutException deriving (Show, Typeable)
+data TimeoutException = TimeoutException String deriving (Show, Typeable)
 instance Exception TimeoutException
 
 connect :: (Host -> CC.PortID -> Maybe Int -> IO CC.ConnectionContext) -> [CMD.CommandInfo] -> MVar ShardMap -> Maybe Int -> Bool -> ([NodeConnection] -> IO ShardMap) -> IO Connection
@@ -266,10 +267,10 @@ evaluatePipeline shardMapVar refreshShardmapAction conn requests = do
         resps <- concat <$>
           mapM (\(resp, (cc, r)) -> case resp of
               Right v -> return v
-              Left (e :: SomeException) ->
-                case displayException e of
-                  "TimeoutException" -> throwIO TimeoutException
-                  _ -> executeRequests (getRandomConnection cc conn) r
+              Left (err :: SomeException) ->
+                if isPrefixOf "TimeoutException" (displayException err)
+                  then throwIO err
+                  else executeRequests (getRandomConnection cc conn) r
             ) (zip eresps requestsByNode)
         -- check for any moved in both responses and continue the flow.
         when (any (moved . rawResponse) resps) refreshShardMapVar
@@ -457,7 +458,7 @@ requestNode (NodeConnection ctx lastRecvRef _) requests = do
     eresp <- race requestNodeImpl (threadDelay envTimeout)
     case eresp of
       Left e -> return e
-      Right _ -> putStrLn "timeout happened" *> throwIO TimeoutException
+      Right _ -> putStrLn "timeout happened" *> throwIO (TimeoutException "Request Timeout")
 
     where
     requestNodeImpl :: IO [Reply]

src/Database/Redis/Connection.hs

@@ -20,6 +20,10 @@ import qualified Data.Time as Time
 import Network.TLS (ClientParams)
 import qualified Network.Socket as NS
 import qualified Data.HashMap.Strict as HM
+import System.Random (randomRIO)
+import System.Environment (lookupEnv)
+import Data.Maybe (fromMaybe)
+import Text.Read (readMaybe)
 
 import qualified Database.Redis.ProtocolPipelining as PP
 import Database.Redis.Core(Redis, runRedisInternal, runRedisClusteredInternal)
@@ -276,19 +280,22 @@ refreshShardMap (Cluster.Connection nodeConns _ _ _ _) =
 
 refreshShardMapWithNodeConn :: [Cluster.NodeConnection] -> IO ShardMap
 refreshShardMapWithNodeConn [] = throwIO $ ClusterConnectError (Error "Couldn't refresh shardMap due to connection error")
-refreshShardMapWithNodeConn ((Cluster.NodeConnection ctx _ _) : xs) = do
+refreshShardMapWithNodeConn nodeConnsList = do
+    selectedIdx <- randomRIO (0, (length nodeConnsList) - 1)
+    let (Cluster.NodeConnection ctx _ _) = nodeConnsList !! selectedIdx
     pipelineConn <- PP.fromCtx ctx
-    raceResult <- race (threadDelay (10^(3 :: Int))) (try $ refreshShardMapWithConn pipelineConn True) -- racing with delay of 1 ms 
+    envTimeout <- fromMaybe (10 ^ (3 :: Int)) . (>>= readMaybe) <$> lookupEnv "REDIS_CLUSTER_SLOTS_TIMEOUT"
+    raceResult <- race (threadDelay envTimeout) (try $ refreshShardMapWithConn pipelineConn True) -- racing with delay of default 1 ms 
     case raceResult of
         Left () -> do
             print $ "TimeoutForConnection " <> show ctx 
-            refreshShardMapWithNodeConn xs
+            throwIO $ Cluster.TimeoutException "ClusterSlots Timeout"
         Right eiShardMapResp -> 
             case eiShardMapResp of
                 Right shardMap -> pure shardMap 
                 Left (err :: SomeException) -> do
                     print $ "ShardMapRefreshError-" <> show err 
-                    refreshShardMapWithNodeConn xs
+                    throwIO $ ClusterConnectError (Error "Couldn't refresh shardMap due to connection error")
 
 refreshShardMapWithConn :: PP.Connection -> Bool -> IO ShardMap
 refreshShardMapWithConn pipelineConn _ = do