Repository: redis/redis-py
Pull request #2816 (2023-06-28 11:11:42)

Save a reference to created async tasks, to avoid tasks potentially disappearing
> Save a reference to the result of this function, to avoid a task disappearing mid-execution. The event loop only keeps weak references to tasks. A task that isn’t referenced elsewhere may get garbage collected at any time, even before it’s done.
>
> [python docs](https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task)

### Pull Request check-list

_Please make sure to review and check all of these items:_

- [ ] Does `$ tox` pass with this change (including linting)?
- [ ] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [ ] Is the new or changed code fully tested?
- [ ] Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
- [ ] Is there an example added to the examples folder (if applicable)?
- [ ] Was the change added to CHANGES file?

_NOTE: these things are not required to open a PR and can be done
afterwards / while the PR is open._

### Description of change

Please review the link: https://textual.textualize.io/blog/2023/02/11/the-heisenbug-lurking-in-your-async-code/

codecov-commenter: ## [Codecov](https://app.codecov.io/gh/redis/redis-py/pull/2816?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) Report
Patch coverage has no change and project coverage change: **`-14.21`** :warning:
> Comparison is base [(`9f50357`)](https://app.codecov.io/gh/redis/redis-py/commit/9f503578d1ffed20d63e8023bcd8a7dccd15ecc5?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) 92.33% compared to head [(`0c62a77`)](https://app.codecov.io/gh/redis/redis-py/pull/2816?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) 78.13%.

:exclamation: Your organization is not using the GitHub App Integration. As a result you may experience degraded service beginning May 15th. Please [install the Github App Integration](https://github.com/apps/codecov) for your organization. [Read more](https://about.codecov.io/blog/codecov-is-updating-its-github-integration/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff             @@
##           master    #2816       +/-   ##
===========================================
- Coverage   92.33%   78.13%   -14.21%     
===========================================
  Files         119      119               
  Lines       30522    30497       -25     
===========================================
- Hits        28184    23830     -4354     
- Misses       2338     6667     +4329     
```


| [Impacted Files](https://app.codecov.io/gh/redis/redis-py/pull/2816?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | Coverage Δ | |
|---|---|---|
| [redis/asyncio/cluster.py](https://app.codecov.io/gh/redis/redis-py/pull/2816?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9jbHVzdGVyLnB5) | `16.93% <0.00%> (-74.84%)` | :arrow_down: |

... and [47 files with indirect coverage changes](https://app.codecov.io/gh/redis/redis-py/pull/2816/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis)


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/redis/redis-py/pull/2816?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).


Patch:
diff --git a/CHANGES b/CHANGES
index 7b3b4c5a..8cfc47db 100644
--- a/CHANGES
+++ b/CHANGES
@@ -1,3 +1,4 @@
+    * Add 'aclose()' methods to async classes, deprecate async close().
     * Fix #2831, add auto_close_connection_pool=True arg to asyncio.Redis.from_url()
     * Fix incorrect redis.asyncio.Cluster type hint for `retry_on_error`
     * Fix dead weakref in sentinel connection causing ReferenceError (#2767)
diff --git a/docs/examples/asyncio_examples.ipynb b/docs/examples/asyncio_examples.ipynb
index a8ee8faf..5eab4db1 100644
--- a/docs/examples/asyncio_examples.ipynb
+++ b/docs/examples/asyncio_examples.ipynb
@@ -15,7 +15,7 @@
     "\n",
     "## Connecting and Disconnecting\n",
     "\n",
-    "Utilizing asyncio Redis requires an explicit disconnect of the connection since there is no asyncio deconstructor magic method. By default, a connection pool is created on `redis.Redis()` and attached to this `Redis` instance. The connection pool closes automatically on the call to `Redis.close` which disconnects all connections."
+    "Utilizing asyncio Redis requires an explicit disconnect of the connection since there is no asyncio deconstructor magic method. By default, a connection pool is created on `redis.Redis()` and attached to this `Redis` instance. The connection pool closes automatically on the call to `Redis.aclose` which disconnects all connections."
    ]
   },
   {
@@ -39,9 +39,9 @@
    "source": [
     "import redis.asyncio as redis\n",
     "\n",
-    "connection = redis.Redis()\n",
-    "print(f\"Ping successful: {await connection.ping()}\")\n",
-    "await connection.close()"
+    "client = redis.Redis()\n",
+    "print(f\"Ping successful: {await client.ping()}\")\n",
+    "await client.aclose()"
    ]
   },
   {
@@ -60,8 +60,8 @@
     "import redis.asyncio as redis\n",
     "\n",
     "pool = redis.ConnectionPool.from_url(\"redis://localhost\")\n",
-    "connection = redis.Redis.from_pool(pool)\n",
-    "await connection.close()"
+    "client = redis.Redis.from_pool(pool)\n",
+    "await client.close()"
    ]
   },
   {
@@ -91,11 +91,11 @@
     "import redis.asyncio as redis\n",
     "\n",
     "pool = redis.ConnectionPool.from_url(\"redis://localhost\")\n",
-    "connection1 = redis.Redis(connection_pool=pool)\n",
-    "connection2 = redis.Redis(connection_pool=pool)\n",
-    "await connection1.close()\n",
-    "await connection2.close()\n",
-    "await pool.disconnect()"
+    "client1 = redis.Redis(connection_pool=pool)\n",
+    "client2 = redis.Redis(connection_pool=pool)\n",
+    "await client1.aclose()\n",
+    "await client2.aclose()\n",
+    "await pool.aclose()"
    ]
   },
   {
@@ -113,9 +113,9 @@
    "source": [
     "import redis.asyncio as redis\n",
     "\n",
-    "connection = redis.Redis(protocol=3)\n",
-    "await connection.close()\n",
-    "await connection.ping()"
+    "client = redis.Redis(protocol=3)\n",
+    "await client.aclose()\n",
+    "await client.ping()"
    ]
   },
   {
diff --git a/redis/_parsers/base.py b/redis/_parsers/base.py
index f77296df..5de04c0f 100644
--- a/redis/_parsers/base.py
+++ b/redis/_parsers/base.py
@@ -138,12 +138,6 @@ class AsyncBaseParser(BaseParser):
         self._stream: Optional[StreamReader] = None
         self._read_size = socket_read_size
 
-    def __del__(self):
-        try:
-            self.on_disconnect()
-        except Exception:
-            pass
-
     async def can_read_destructive(self) -> bool:
         raise NotImplementedError()
 
diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index cce1b581..e4d2e776 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -14,7 +14,6 @@ from typing import (
     List,
     Mapping,
     MutableMapping,
-    NoReturn,
     Optional,
     Set,
     Tuple,
@@ -65,6 +64,7 @@ from redis.typing import ChannelT, EncodableT, KeyT
 from redis.utils import (
     HIREDIS_AVAILABLE,
     _set_info_logger,
+    deprecated_function,
     get_lib_version,
     safe_str,
     str_if_bytes,
@@ -227,7 +227,6 @@ class Redis(
         lib_version: Optional[str] = get_lib_version(),
         username: Optional[str] = None,
         retry: Optional[Retry] = None,
-        # deprecated. create a pool and use connection_pool instead
         auto_close_connection_pool: Optional[bool] = None,
         redis_connect_func=None,
         credential_provider: Optional[CredentialProvider] = None,
@@ -241,7 +240,9 @@ class Redis(
         To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
         kwargs: Dict[str, Any]
-
+        # auto_close_connection_pool only has an effect if connection_pool is
+        # None. It is assumed that if connection_pool is not None, the user
+        # wants to manage the connection pool themselves.
         if auto_close_connection_pool is not None:
             warnings.warn(
                 DeprecationWarning(
@@ -527,19 +528,26 @@ class Redis(
         return await self.initialize()
 
     async def __aexit__(self, exc_type, exc_value, traceback):
-        await self.close()
+        await self.aclose()
 
     _DEL_MESSAGE = "Unclosed Redis client"
 
-    def __del__(self, _warnings: Any = warnings) -> None:
+    # passing _warnings and _grl as argument default since they may be gone
+    # by the time __del__ is called at shutdown
+    def __del__(
+        self,
+        _warn: Any = warnings.warn,
+        _grl: Any = asyncio.get_running_loop,
+    ) -> None:
         if hasattr(self, "connection") and (self.connection is not None):
-            _warnings.warn(
-                f"Unclosed client session {self!r}", ResourceWarning, source=self
-            )
-            context = {"client": self, "message": self._DEL_MESSAGE}
-            asyncio.get_running_loop().call_exception_handler(context)
+            _warn(f"Unclosed client session {self!r}", ResourceWarning, source=self)
+            try:
+                context = {"client": self, "message": self._DEL_MESSAGE}
+                _grl().call_exception_handler(context)
+            except RuntimeError:
+                pass
 
-    async def close(self, close_connection_pool: Optional[bool] = None) -> None:
+    async def aclose(self, close_connection_pool: Optional[bool] = None) -> None:
         """
         Closes Redis client connection
 
@@ -557,6 +565,13 @@ class Redis(
         ):
             await self.connection_pool.disconnect()
 
+    @deprecated_function(version="5.0.0", reason="Use aclose() instead", name="close")
+    async def close(self, close_connection_pool: Optional[bool] = None) -> None:
+        """
+        Alias for aclose(), for backwards compatibility
+        """
+        await self.aclose(close_connection_pool)
+
     async def _send_command_parse_response(self, conn, command_name, *args, **options):
         """
         Send a command and parse the response
@@ -645,7 +660,7 @@ class Monitor:
     listen() method yields commands from monitor.
     """
 
-    monitor_re = re.compile(r"\[(\d+) (.*)\] (.*)")
+    monitor_re = re.compile(r"\[(\d+) (.*?)\] (.*)")
     command_re = re.compile(r'"(.*?)(?<!\\)"')
 
     def __init__(self, connection_pool: ConnectionPool):
@@ -764,17 +779,22 @@ class PubSub:
         return self
 
     async def __aexit__(self, exc_type, exc_value, traceback):
-        await self.reset()
+        await self.aclose()
 
     def __del__(self):
         if self.connection:
-            self.connection.clear_connect_callbacks()
+            self.connection._deregister_connect_callback(self.on_connect)
 
-    async def reset(self):
+    async def aclose(self):
+        # In case a connection property does not yet exist
+        # (due to a crash earlier in the Redis() constructor), return
+        # immediately as there is nothing to clean-up.
+        if not hasattr(self, "connection"):
+            return
         async with self._lock:
             if self.connection:
                 await self.connection.disconnect()
-                self.connection.clear_connect_callbacks()
+                self.connection._deregister_connect_callback(self.on_connect)
                 await self.connection_pool.release(self.connection)
                 self.connection = None
             self.channels = {}
@@ -782,13 +802,15 @@ class PubSub:
             self.patterns = {}
             self.pending_unsubscribe_patterns = set()
 
-    def close(self) -> Awaitable[NoReturn]:
-        # In case a connection property does not yet exist
-        # (due to a crash earlier in the Redis() constructor), return
-        # immediately as there is nothing to clean-up.
-        if not hasattr(self, "connection"):
-            return
-        return self.reset()
+    @deprecated_function(version="5.0.0", reason="Use aclose() instead", name="close")
+    async def close(self) -> None:
+        """Alias for aclose(), for backwards compatibility"""
+        await self.aclose()
+
+    @deprecated_function(version="5.0.0", reason="Use aclose() instead", name="reset")
+    async def reset(self) -> None:
+        """Alias for aclose(), for backwards compatibility"""
+        await self.aclose()
 
     async def on_connect(self, connection: Connection):
         """Re-subscribe to any channels and patterns previously subscribed to"""
@@ -835,7 +857,7 @@ class PubSub:
             )
             # register a callback that re-subscribes to any channels we
             # were listening to when we were disconnected
-            self.connection.register_connect_callback(self.on_connect)
+            self.connection._register_connect_callback(self.on_connect)
         else:
             await self.connection.connect()
         if self.push_handler_func is not None and not HIREDIS_AVAILABLE:
@@ -1232,6 +1254,10 @@ class Pipeline(Redis):  # lgtm [py/init-calls-subclass]
             await self.connection_pool.release(self.connection)
             self.connection = None
 
+    async def aclose(self) -> None:
+        """Alias for reset(), a standard method name for cleanup"""
+        await self.reset()
+
     def multi(self):
         """
         Start a transactional block of the pipeline after WATCH commands
@@ -1264,14 +1290,14 @@ class Pipeline(Redis):  # lgtm [py/init-calls-subclass]
         # valid since this connection has died. raise a WatchError, which
         # indicates the user should retry this transaction.
         if self.watching:
-            await self.reset()
+            await self.aclose()
             raise WatchError(
                 "A ConnectionError occurred on while watching one or more keys"
             )
         # if retry_on_timeout is not set, or the error is not
         # a TimeoutError, raise it
         if not (conn.retry_on_timeout and isinstance(error, TimeoutError)):
-            await self.reset()
+            await self.aclose()
             raise
 
     async def immediate_execute_command(self, *args, **options):
diff --git a/redis/asyncio/cluster.py b/redis/asyncio/cluster.py
index 65e1bd02..636144a9 100644
--- a/redis/asyncio/cluster.py
+++ b/redis/asyncio/cluster.py
@@ -62,7 +62,13 @@ from redis.exceptions import (
     TryAgainError,
 )
 from redis.typing import AnyKeyT, EncodableT, KeyT
-from redis.utils import dict_merge, get_lib_version, safe_str, str_if_bytes
+from redis.utils import (
+    deprecated_function,
+    dict_merge,
+    get_lib_version,
+    safe_str,
+    str_if_bytes,
+)
 
 TargetNodesT = TypeVar(
     "TargetNodesT", str, "ClusterNode", List["ClusterNode"], Dict[Any, "ClusterNode"]
@@ -395,12 +401,12 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
                         )
                         self._initialize = False
                     except BaseException:
-                        await self.nodes_manager.close()
-                        await self.nodes_manager.close("startup_nodes")
+                        await self.nodes_manager.aclose()
+                        await self.nodes_manager.aclose("startup_nodes")
                         raise
         return self
 
-    async def close(self) -> None:
+    async def aclose(self) -> None:
         """Close all connections & client if initialized."""
         if not self._initialize:
             if not self._lock:
@@ -408,28 +414,37 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
             async with self._lock:
                 if not self._initialize:
                     self._initialize = True
-                    await self.nodes_manager.close()
-                    await self.nodes_manager.close("startup_nodes")
+                    await self.nodes_manager.aclose()
+                    await self.nodes_manager.aclose("startup_nodes")
+
+    @deprecated_function(version="5.0.0", reason="Use aclose() instead", name="close")
+    async def close(self) -> None:
+        """alias for aclose() for backwards compatibility"""
+        await self.aclose()
 
     async def __aenter__(self) -> "RedisCluster":
         return await self.initialize()
 
     async def __aexit__(self, exc_type: None, exc_value: None, traceback: None) -> None:
-        await self.close()
+        await self.aclose()
 
     def __await__(self) -> Generator[Any, None, "RedisCluster"]:
         return self.initialize().__await__()
 
     _DEL_MESSAGE = "Unclosed RedisCluster client"
 
-    def __del__(self) -> None:
+    def __del__(
+        self,
+        _warn: Any = warnings.warn,
+        _grl: Any = asyncio.get_running_loop,
+    ) -> None:
         if hasattr(self, "_initialize") and not self._initialize:
-            warnings.warn(f"{self._DEL_MESSAGE} {self!r}", ResourceWarning, source=self)
+            _warn(f"{self._DEL_MESSAGE} {self!r}", ResourceWarning, source=self)
             try:
                 context = {"client": self, "message": self._DEL_MESSAGE}
-                asyncio.get_running_loop().call_exception_handler(context)
+                _grl().call_exception_handler(context)
             except RuntimeError:
-                ...
+                pass
 
     async def on_connect(self, connection: Connection) -> None:
         await connection.on_connect()
@@ -767,13 +782,13 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
                 self.nodes_manager.startup_nodes.pop(target_node.name, None)
                 # Hard force of reinitialize of the node/slots setup
                 # and try again with the new setup
-                await self.close()
+                await self.aclose()
                 raise
             except ClusterDownError:
                 # ClusterDownError can occur during a failover and to get
                 # self-healed, we will try to reinitialize the cluster layout
                 # and retry executing the command
-                await self.close()
+                await self.aclose()
                 await asyncio.sleep(0.25)
                 raise
             except MovedError as e:
@@ -790,7 +805,7 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
                     self.reinitialize_steps
                     and self.reinitialize_counter % self.reinitialize_steps == 0
                 ):
-                    await self.close()
+                    await self.aclose()
                     # Reset the counter
                     self.reinitialize_counter = 0
                 else:
@@ -958,17 +973,20 @@ class ClusterNode:
 
     _DEL_MESSAGE = "Unclosed ClusterNode object"
 
-    def __del__(self) -> None:
+    def __del__(
+        self,
+        _warn: Any = warnings.warn,
+        _grl: Any = asyncio.get_running_loop,
+    ) -> None:
         for connection in self._connections:
             if connection.is_connected:
-                warnings.warn(
-                    f"{self._DEL_MESSAGE} {self!r}", ResourceWarning, source=self
-                )
+                _warn(f"{self._DEL_MESSAGE} {self!r}", ResourceWarning, source=self)
+
                 try:
                     context = {"client": self, "message": self._DEL_MESSAGE}
-                    asyncio.get_running_loop().call_exception_handler(context)
+                    _grl().call_exception_handler(context)
                 except RuntimeError:
-                    ...
+                    pass
                 break
 
     async def disconnect(self) -> None:
@@ -1117,13 +1135,13 @@ class NodesManager:
         if remove_old:
             for name in list(old.keys()):
                 if name not in new:
-                    asyncio.create_task(old.pop(name).disconnect())
+                    task = asyncio.create_task(old.pop(name).disconnect())  # noqa
 
         for name, node in new.items():
             if name in old:
                 if old[name] is node:
                     continue
-                asyncio.create_task(old[name].disconnect())
+                task = asyncio.create_task(old[name].disconnect())  # noqa
             old[name] = node
 
     def _update_moved_slots(self) -> None:
@@ -1323,7 +1341,7 @@ class NodesManager:
         # If initialize was called after a MovedError, clear it
         self._moved_exception = None
 
-    async def close(self, attr: str = "nodes_cache") -> None:
+    async def aclose(self, attr: str = "nodes_cache") -> None:
         self.default_node = None
         await asyncio.gather(
             *(
@@ -1471,7 +1489,7 @@ class ClusterPipeline(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterComm
                     if type(e) in self.__class__.ERRORS_ALLOW_RETRY:
                         # Try again with the new cluster setup.
                         exception = e
-                        await self._client.close()
+                        await self._client.aclose()
                         await asyncio.sleep(0.25)
                     else:
                         # All other errors should be raised.
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index 4f20e6ec..f36b4bf7 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -216,11 +216,16 @@ class AbstractConnection:
     def is_connected(self):
         return self._reader is not None and self._writer is not None
 
-    def register_connect_callback(self, callback):
-        self._connect_callbacks.append(weakref.WeakMethod(callback))
+    def _register_connect_callback(self, callback):
+        wm = weakref.WeakMethod(callback)
+        if wm not in self._connect_callbacks:
+            self._connect_callbacks.append(wm)
 
-    def clear_connect_callbacks(self):
-        self._connect_callbacks = []
+    def _deregister_connect_callback(self, callback):
+        try:
+            self._connect_callbacks.remove(weakref.WeakMethod(callback))
+        except ValueError:
+            pass
 
     def set_parser(self, parser_class: Type[BaseParser]) -> None:
         """
@@ -263,6 +268,8 @@ class AbstractConnection:
 
         # run any user callbacks. right now the only internal callback
         # is for pubsub channel/pattern resubscription
+        # first, remove any dead weakrefs
+        self._connect_callbacks = [ref for ref in self._connect_callbacks if ref()]
         for ref in self._connect_callbacks:
             callback = ref()
             task = callback(self)
@@ -1095,6 +1102,10 @@ class ConnectionPool:
         if exc:
             raise exc
 
+    async def aclose(self) -> None:
+        """Close the pool, disconnecting all connections"""
+        await self.disconnect()
+
     def set_retry(self, retry: "Retry") -> None:
         for conn in self._available_connections:
             conn.retry = retry
diff --git a/redis/client.py b/redis/client.py
index 017f8ef1..cf6dbf1e 100755
--- a/redis/client.py
+++ b/redis/client.py
@@ -571,7 +571,7 @@ class Monitor:
     listen() method yields commands from monitor.
     """
 
-    monitor_re = re.compile(r"\[(\d+) (.*)\] (.*)")
+    monitor_re = re.compile(r"\[(\d+) (.*?)\] (.*)")
     command_re = re.compile(r'"(.*?)(?<!\\)"')
 
     def __init__(self, connection_pool):
@@ -690,7 +690,7 @@ class PubSub:
     def reset(self):
         if self.connection:
             self.connection.disconnect()
-            self.connection.clear_connect_callbacks()
+            self.connection._deregister_connect_callback(self.on_connect)
             self.connection_pool.release(self.connection)
             self.connection = None
         self.health_check_response_counter = 0
@@ -748,7 +748,7 @@ class PubSub:
             )
             # register a callback that re-subscribes to any channels we
             # were listening to when we were disconnected
-            self.connection.register_connect_callback(self.on_connect)
+            self.connection._register_connect_callback(self.on_connect)
             if self.push_handler_func is not None and not HIREDIS_AVAILABLE:
                 self.connection._parser.set_push_handler(self.push_handler_func)
         connection = self.connection
@@ -1217,6 +1217,10 @@ class Pipeline(Redis):
             self.connection_pool.release(self.connection)
             self.connection = None
 
+    def close(self):
+        """Close the pipeline"""
+        self.reset()
+
     def multi(self):
         """
         Start a transactional block of the pipeline after WATCH commands
diff --git a/redis/cluster.py b/redis/cluster.py
index 03478932..ee3e1a86 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -295,6 +295,10 @@ class AbstractRedisCluster:
                 "LATENCY HISTORY",
                 "LATENCY LATEST",
                 "LATENCY RESET",
+                "MODULE LIST",
+                "MODULE LOAD",
+                "MODULE UNLOAD",
+                "MODULE LOADEX",
             ],
             DEFAULT_NODE,
         ),
@@ -1771,7 +1775,7 @@ class ClusterPubSub(PubSub):
             )
             # register a callback that re-subscribes to any channels we
             # were listening to when we were disconnected
-            self.connection.register_connect_callback(self.on_connect)
+            self.connection._register_connect_callback(self.on_connect)
             if self.push_handler_func is not None and not HIREDIS_AVAILABLE:
                 self.connection._parser.set_push_handler(self.push_handler_func)
         connection = self.connection
diff --git a/redis/commands/cluster.py b/redis/commands/cluster.py
index 691cab3d..14b87414 100644
--- a/redis/commands/cluster.py
+++ b/redis/commands/cluster.py
@@ -32,11 +32,13 @@ from .core import (
     AsyncFunctionCommands,
     AsyncGearsCommands,
     AsyncManagementCommands,
+    AsyncModuleCommands,
     AsyncScriptCommands,
     DataAccessCommands,
     FunctionCommands,
     GearsCommands,
     ManagementCommands,
+    ModuleCommands,
     PubSubCommands,
     ResponseT,
     ScriptCommands,
@@ -873,6 +875,7 @@ class RedisClusterCommands(
     ScriptCommands,
     FunctionCommands,
     GearsCommands,
+    ModuleCommands,
     RedisModuleCommands,
 ):
     """
@@ -903,6 +906,7 @@ class AsyncRedisClusterCommands(
     AsyncScriptCommands,
     AsyncFunctionCommands,
     AsyncGearsCommands,
+    AsyncModuleCommands,
 ):
     """
     A class for all Redis Cluster commands
diff --git a/redis/commands/core.py b/redis/commands/core.py
index 031781d7..9d81e977 100644
--- a/redis/commands/core.py
+++ b/redis/commands/core.py
@@ -2823,13 +2823,13 @@ class ListCommands(CommandsProtocol):
         """
         return self.execute_command("RPUSH", name, *values)
 
-    def rpushx(self, name: str, value: str) -> Union[Awaitable[int], int]:
+    def rpushx(self, name: str, *values: str) -> Union[Awaitable[int], int]:
         """
         Push ``value`` onto the tail of the list ``name`` if ``name`` exists
 
         For more information see https://redis.io/commands/rpushx
         """
-        return self.execute_command("RPUSHX", name, value)
+        return self.execute_command("RPUSHX", name, *values)
 
     def lpos(
         self,
diff --git a/redis/connection.py b/redis/connection.py
index c6a22aae..f5266d7d 100644
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -237,11 +237,16 @@ class AbstractConnection:
         else:
             return PythonRespSerializer(self._buffer_cutoff, self.encoder.encode)
 
-    def register_connect_callback(self, callback):
-        self._connect_callbacks.append(weakref.WeakMethod(callback))
+    def _register_connect_callback(self, callback):
+        wm = weakref.WeakMethod(callback)
+        if wm not in self._connect_callbacks:
+            self._connect_callbacks.append(wm)
 
-    def clear_connect_callbacks(self):
-        self._connect_callbacks = []
+    def _deregister_connect_callback(self, callback):
+        try:
+            self._connect_callbacks.remove(weakref.WeakMethod(callback))
+        except ValueError:
+            pass
 
     def set_parser(self, parser_class):
         """
@@ -279,6 +284,8 @@ class AbstractConnection:
 
         # run any user callbacks. right now the only internal callback
         # is for pubsub channel/pattern resubscription
+        # first, remove any dead weakrefs
+        self._connect_callbacks = [ref for ref in self._connect_callbacks if ref()]
         for ref in self._connect_callbacks:
             callback = ref()
             if callback:
@@ -1154,6 +1161,10 @@ class ConnectionPool:
             for connection in connections:
                 connection.disconnect()
 
+    def close(self) -> None:
+        """Close the pool, disconnecting all connections"""
+        self.disconnect()
+
     def set_retry(self, retry: "Retry") -> None:
         self.connection_kwargs.update({"retry": retry})
         for conn in self._available_connections:


Test patch:
diff --git a/tests/test_asyncio/compat.py b/tests/test_asyncio/compat.py
index 5edcd4ae..4a9778b7 100644
--- a/tests/test_asyncio/compat.py
+++ b/tests/test_asyncio/compat.py
@@ -6,6 +6,18 @@ try:
 except AttributeError:
     import mock
 
+try:
+    from contextlib import aclosing
+except ImportError:
+    import contextlib
+
+    @contextlib.asynccontextmanager
+    async def aclosing(thing):
+        try:
+            yield thing
+        finally:
+            await thing.aclose()
+
 
 def create_task(coroutine):
     return asyncio.create_task(coroutine)
diff --git a/tests/test_asyncio/conftest.py b/tests/test_asyncio/conftest.py
index 10ab4732..5d9e0b4f 100644
--- a/tests/test_asyncio/conftest.py
+++ b/tests/test_asyncio/conftest.py
@@ -100,7 +100,7 @@ async def create_redis(request):
                         # handle cases where a test disconnected a client
                         # just manually retry the flushdb
                         await client.flushdb()
-                await client.close()
+                await client.aclose()
                 await client.connection_pool.disconnect()
             else:
                 if flushdb:
@@ -110,7 +110,7 @@ async def create_redis(request):
                         # handle cases where a test disconnected a client
                         # just manually retry the flushdb
                         await client.flushdb(target_nodes="primaries")
-                await client.close()
+                await client.aclose()
 
         teardown_clients.append(teardown)
         return client
diff --git a/tests/test_asyncio/test_cluster.py b/tests/test_asyncio/test_cluster.py
index 332101ed..e6cf2e4c 100644
--- a/tests/test_asyncio/test_cluster.py
+++ b/tests/test_asyncio/test_cluster.py
@@ -38,7 +38,7 @@ from tests.conftest import (
 )
 
 from ..ssl_utils import get_ssl_filename
-from .compat import mock
+from .compat import aclosing, mock
 
 pytestmark = pytest.mark.onlycluster
 
@@ -270,7 +270,38 @@ class TestRedisClusterObj:
         cluster = await get_mocked_redis_client(host=default_host, port=default_port)
         assert cluster.get_node(host=default_host, port=default_port) is not None
 
-        await cluster.close()
+        await cluster.aclose()
+
+    async def test_aclosing(self) -> None:
+        cluster = await get_mocked_redis_client(host=default_host, port=default_port)
+        called = 0
+
+        async def mock_aclose():
+            nonlocal called
+            called += 1
+
+        with mock.patch.object(cluster, "aclose", mock_aclose):
+            async with aclosing(cluster):
+                pass
+            assert called == 1
+        await cluster.aclose()
+
+    async def test_close_is_aclose(self) -> None:
+        """
+        Test that it is possible to use host & port arguments as startup node
+        args
+        """
+        cluster = await get_mocked_redis_client(host=default_host, port=default_port)
+        called = 0
+
+        async def mock_aclose():
+            nonlocal called
+            called += 1
+
+        with mock.patch.object(cluster, "aclose", mock_aclose):
+            await cluster.close()
+            assert called == 1
+        await cluster.aclose()
 
     async def test_startup_nodes(self) -> None:
         """
@@ -289,7 +320,7 @@ class TestRedisClusterObj:
             and cluster.get_node(host=default_host, port=port_2) is not None
         )
 
-        await cluster.close()
+        await cluster.aclose()
 
         startup_node = ClusterNode("127.0.0.1", 16379)
         async with RedisCluster(startup_nodes=[startup_node], client_name="test") as rc:
@@ -417,7 +448,7 @@ class TestRedisClusterObj:
                     )
                 )
 
-        await rc.close()
+        await rc.aclose()
 
     async def test_execute_command_errors(self, r: RedisCluster) -> None:
         """
@@ -461,7 +492,7 @@ class TestRedisClusterObj:
             conn = primary._free.pop()
             assert conn.read_response.called is not True
 
-        await r.close()
+        await r.aclose()
 
     async def test_execute_command_node_flag_all_nodes(self, r: RedisCluster) -> None:
         """
@@ -690,7 +721,7 @@ class TestRedisClusterObj:
                 await read_cluster.get("foo")
                 mocks["send_command"].assert_has_calls([mock.call("READONLY")])
 
-                await read_cluster.close()
+                await read_cluster.aclose()
 
     async def test_keyslot(self, r: RedisCluster) -> None:
         """
@@ -762,7 +793,7 @@ class TestRedisClusterObj:
                 await rc.get("bar")
                 assert execute_command.failed_calls == rc.cluster_error_retry_attempts
 
-            await rc.close()
+            await rc.aclose()
 
     async def test_set_default_node_success(self, r: RedisCluster) -> None:
         """
@@ -843,7 +874,7 @@ class TestRedisClusterObj:
                 *(rc.echo("i", target_nodes=RedisCluster.ALL_NODES) for i in range(100))
             )
         )
-        await rc.close()
+        await rc.aclose()
 
     def test_replace_cluster_node(self, r: RedisCluster) -> None:
         prev_default_node = r.get_default_node()
@@ -901,7 +932,7 @@ class TestRedisClusterObj:
                 assert await r.set("byte_string", b"giraffe")
                 assert await r.get("byte_string") == b"giraffe"
             finally:
-                await r.close()
+                await r.aclose()
         finally:
             await asyncio.gather(*[p.aclose() for p in proxies])
 
@@ -1002,7 +1033,7 @@ class TestClusterRedisCommands:
         url = request.config.getoption("--redis-url")
         r = RedisCluster.from_url(url)
         assert 0 == await r.exists("a", "b", "c")
-        await r.close()
+        await r.aclose()
 
     @skip_if_redis_enterprise()
     async def test_cluster_myid(self, r: RedisCluster) -> None:
@@ -1065,7 +1096,7 @@ class TestClusterRedisCommands:
         assert node0._free.pop().read_response.called
         assert node1._free.pop().read_response.called
 
-        await r.close()
+        await r.aclose()
 
     @skip_if_server_version_lt("7.0.0")
     @skip_if_redis_enterprise()
@@ -1076,7 +1107,7 @@ class TestClusterRedisCommands:
         await r.cluster_addslots(node, 1, 2, 3, 4, 5)
         assert await r.cluster_delslotsrange(1, 5)
         assert node._free.pop().read_response.called
-        await r.close()
+        await r.aclose()
 
     @skip_if_redis_enterprise()
     async def test_cluster_failover(self, r: RedisCluster) -> None:
@@ -1286,7 +1317,7 @@ class TestClusterRedisCommands:
         for replica in r.get_replicas():
             assert replica._free.pop().read_response.called
 
-        await r.close()
+        await r.aclose()
 
     @skip_if_redis_enterprise()
     async def test_readwrite(self) -> None:
@@ -1299,7 +1330,7 @@ class TestClusterRedisCommands:
         for replica in r.get_replicas():
             assert replica._free.pop().read_response.called
 
-        await r.close()
+        await r.aclose()
 
     @skip_if_redis_enterprise()
     async def test_bgsave(self, r: RedisCluster) -> None:
@@ -1524,7 +1555,7 @@ class TestClusterRedisCommands:
         ]
         assert len(clients) == 1
         assert clients[0].get("name") == "redis-py-c1"
-        await r2.close()
+        await r2.aclose()
 
     @skip_if_server_version_lt("2.6.0")
     async def test_cluster_bitop_not_empty_string(self, r: RedisCluster) -> None:
@@ -2302,7 +2333,7 @@ class TestClusterRedisCommands:
 
         await r.acl_deluser(username, target_nodes="primaries")
 
-        await user_client.close()
+        await user_client.aclose()
 
 
 class TestNodesManager:
@@ -2359,7 +2390,7 @@ class TestNodesManager:
                 cluster_slots=cluster_slots,
                 require_full_coverage=True,
             )
-            await rc.close()
+            await rc.aclose()
         assert str(ex.value).startswith(
             "All slots are not covered after query all startup_nodes."
         )
@@ -2385,7 +2416,7 @@ class TestNodesManager:
 
         assert 5460 not in rc.nodes_manager.slots_cache
 
-        await rc.close()
+        await rc.aclose()
 
     async def test_init_slots_cache(self) -> None:
         """
@@ -2416,7 +2447,7 @@ class TestNodesManager:
 
         assert len(n_manager.nodes_cache) == 6
 
-        await rc.close()
+        await rc.aclose()
 
     async def test_init_slots_cache_cluster_mode_disabled(self) -> None:
         """
@@ -2427,7 +2458,7 @@ class TestNodesManager:
             rc = await get_mocked_redis_client(
                 host=default_host, port=default_port, cluster_enabled=False
             )
-            await rc.close()
+            await rc.aclose()
         assert "Cluster mode is not enabled on this node" in str(e.value)
 
     async def test_empty_startup_nodes(self) -> None:
@@ -2514,7 +2545,7 @@ class TestNodesManager:
         for i in range(0, REDIS_CLUSTER_HASH_SLOTS):
             assert n.slots_cache[i] == [n_node]
 
-        await rc.close()
+        await rc.aclose()
 
     async def test_init_with_down_node(self) -> None:
         """
diff --git a/tests/test_asyncio/test_commands.py b/tests/test_asyncio/test_commands.py
index 6a3a2eca..9b9852e9 100644
--- a/tests/test_asyncio/test_commands.py
+++ b/tests/test_asyncio/test_commands.py
@@ -3213,7 +3213,6 @@ class TestRedisCommands:
         assert isinstance(await r.memory_usage("foo"), int)
 
     @skip_if_server_version_lt("4.0.0")
-    @pytest.mark.onlynoncluster
     async def test_module_list(self, r: redis.Redis):
         assert isinstance(await r.module_list(), list)
         for x in await r.module_list():
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index a954a40d..28e6b0d9 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -302,7 +302,25 @@ async def test_pool_auto_close(request, from_url):
 
     r1 = await get_redis_connection()
     assert r1.auto_close_connection_pool is True
-    await r1.close()
+    await r1.aclose()
+
+
+async def test_close_is_aclose(request):
+    """Verify close() calls aclose()"""
+    calls = 0
+
+    async def mock_aclose(self):
+        nonlocal calls
+        calls += 1
+
+    url: str = request.config.getoption("--redis-url")
+    r1 = await Redis.from_url(url)
+    with patch.object(r1, "aclose", mock_aclose):
+        await r1.close()
+        assert calls == 1
+
+    with pytest.deprecated_call():
+        await r1.close()
 
 
 async def test_pool_from_url_deprecation(request):
@@ -326,7 +344,7 @@ async def test_pool_auto_close_disable(request):
     r1 = await get_redis_connection()
     assert r1.auto_close_connection_pool is False
     await r1.connection_pool.disconnect()
-    await r1.close()
+    await r1.aclose()
 
 
 @pytest.mark.parametrize("from_url", (True, False), ids=("from_url", "from_args"))
diff --git a/tests/test_asyncio/test_connection_pool.py b/tests/test_asyncio/test_connection_pool.py
index 999c0337..c93fa91a 100644
--- a/tests/test_asyncio/test_connection_pool.py
+++ b/tests/test_asyncio/test_connection_pool.py
@@ -7,7 +7,7 @@ import redis.asyncio as redis
 from redis.asyncio.connection import Connection, to_bool
 from tests.conftest import skip_if_redis_enterprise, skip_if_server_version_lt
 
-from .compat import mock
+from .compat import aclosing, mock
 from .conftest import asynccontextmanager
 from .test_pubsub import wait_for_message
 
@@ -42,7 +42,7 @@ class TestRedisAutoReleaseConnectionPool:
         new_conn = await self.create_two_conn(r)
         assert new_conn != r.connection
         assert self.get_total_connected_connections(r.connection_pool) == 2
-        await r.close()
+        await r.aclose()
         assert self.has_no_connected_connections(r.connection_pool)
 
     async def test_do_not_auto_disconnect_redis_created_pool(self, r2: redis.Redis):
@@ -52,7 +52,7 @@ class TestRedisAutoReleaseConnectionPool:
         )
         new_conn = await self.create_two_conn(r2)
         assert self.get_total_connected_connections(r2.connection_pool) == 2
-        await r2.close()
+        await r2.aclose()
         assert r2.connection_pool._in_use_connections == {new_conn}
         assert new_conn.is_connected
         assert len(r2.connection_pool._available_connections) == 1
@@ -61,7 +61,7 @@ class TestRedisAutoReleaseConnectionPool:
     async def test_auto_release_override_true_manual_created_pool(self, r: redis.Redis):
         assert r.auto_close_connection_pool is True, "This is from the class fixture"
         await self.create_two_conn(r)
-        await r.close()
+        await r.aclose()
         assert self.get_total_connected_connections(r.connection_pool) == 2, (
             "The connection pool should not be disconnected as a manually created "
             "connection pool was passed in in conftest.py"
@@ -72,7 +72,7 @@ class TestRedisAutoReleaseConnectionPool:
     async def test_close_override(self, r: redis.Redis, auto_close_conn_pool):
         r.auto_close_connection_pool = auto_close_conn_pool
         await self.create_two_conn(r)
-        await r.close(close_connection_pool=True)
+        await r.aclose(close_connection_pool=True)
         assert self.has_no_connected_connections(r.connection_pool)
 
     @pytest.mark.parametrize("auto_close_conn_pool", [True, False])
@@ -81,7 +81,7 @@ class TestRedisAutoReleaseConnectionPool:
     ):
         r.auto_close_connection_pool = auto_close_conn_pool
         new_conn = await self.create_two_conn(r)
-        await r.close(close_connection_pool=False)
+        await r.aclose(close_connection_pool=False)
         assert not self.has_no_connected_connections(r.connection_pool)
         assert r.connection_pool._in_use_connections == {new_conn}
         assert r.connection_pool._available_connections[0].is_connected
@@ -135,6 +135,16 @@ class TestConnectionPool:
             assert isinstance(connection, DummyConnection)
             assert connection.kwargs == connection_kwargs
 
+    async def test_aclosing(self):
+        connection_kwargs = {"foo": "bar", "biz": "baz"}
+        pool = redis.ConnectionPool(
+            connection_class=DummyConnection,
+            max_connections=None,
+            **connection_kwargs,
+        )
+        async with aclosing(pool):
+            pass
+
     async def test_multiple_connections(self, master_host):
         connection_kwargs = {"host": master_host[0]}
         async with self.get_pool(connection_kwargs=connection_kwargs) as pool:
diff --git a/tests/test_asyncio/test_pipeline.py b/tests/test_asyncio/test_pipeline.py
index edd2f6d1..0fa12047 100644
--- a/tests/test_asyncio/test_pipeline.py
+++ b/tests/test_asyncio/test_pipeline.py
@@ -2,6 +2,7 @@ import pytest
 import redis
 from tests.conftest import skip_if_server_version_lt
 
+from .compat import aclosing, mock
 from .conftest import wait_for_command
 
 
@@ -286,6 +287,24 @@ class TestPipeline:
             assert unwatch_command is not None
             assert unwatch_command["command"] == "UNWATCH"
 
+    @pytest.mark.onlynoncluster
+    async def test_aclose_is_reset(self, r):
+        async with r.pipeline() as pipe:
+            called = 0
+
+            async def mock_reset():
+                nonlocal called
+                called += 1
+
+            with mock.patch.object(pipe, "reset", mock_reset):
+                await pipe.aclose()
+                assert called == 1
+
+    @pytest.mark.onlynoncluster
+    async def test_aclosing(self, r):
+        async with aclosing(r.pipeline()):
+            pass
+
     @pytest.mark.onlynoncluster
     async def test_transaction_callable(self, r):
         await r.set("a", 1)
diff --git a/tests/test_asyncio/test_pubsub.py b/tests/test_asyncio/test_pubsub.py
index 85857658..8fef34d8 100644
--- a/tests/test_asyncio/test_pubsub.py
+++ b/tests/test_asyncio/test_pubsub.py
@@ -20,7 +20,7 @@ from redis.typing import EncodableT
 from redis.utils import HIREDIS_AVAILABLE
 from tests.conftest import get_protocol_version, skip_if_server_version_lt
 
-from .compat import create_task, mock
+from .compat import aclosing, create_task, mock
 
 
 def with_timeout(t):
@@ -84,9 +84,8 @@ def make_subscribe_test_data(pubsub, type):
 
 @pytest_asyncio.fixture()
 async def pubsub(r: redis.Redis):
-    p = r.pubsub()
-    yield p
-    await p.close()
+    async with r.pubsub() as p:
+        yield p
 
 
 @pytest.mark.onlynoncluster
@@ -217,6 +216,46 @@ class TestPubSubSubscribeUnsubscribe:
         kwargs = make_subscribe_test_data(pubsub, "pattern")
         await self._test_subscribed_property(**kwargs)
 
+    async def test_aclosing(self, r: redis.Redis):
+        p = r.pubsub()
+        async with aclosing(p):
+            assert p.subscribed is False
+            await p.subscribe("foo")
+            assert p.subscribed is True
+        assert p.subscribed is False
+
+    async def test_context_manager(self, r: redis.Redis):
+        p = r.pubsub()
+        async with p:
+            assert p.subscribed is False
+            await p.subscribe("foo")
+            assert p.subscribed is True
+        assert p.subscribed is False
+
+    async def test_close_is_aclose(self, r: redis.Redis):
+        """
+        Test backwards compatible close method
+        """
+        p = r.pubsub()
+        assert p.subscribed is False
+        await p.subscribe("foo")
+        assert p.subscribed is True
+        with pytest.deprecated_call():
+            await p.close()
+        assert p.subscribed is False
+
+    async def test_reset_is_aclose(self, r: redis.Redis):
+        """
+        Test backwards compatible reset method
+        """
+        p = r.pubsub()
+        assert p.subscribed is False
+        await p.subscribe("foo")
+        assert p.subscribed is True
+        with pytest.deprecated_call():
+            await p.reset()
+        assert p.subscribed is False
+
     async def test_ignore_all_subscribe_messages(self, r: redis.Redis):
         p = r.pubsub(ignore_subscribe_messages=True)
 
@@ -233,7 +272,7 @@ class TestPubSubSubscribeUnsubscribe:
             assert p.subscribed is True
             assert await wait_for_message(p) is None
         assert p.subscribed is False
-        await p.close()
+        await p.aclose()
 
     async def test_ignore_individual_subscribe_messages(self, pubsub):
         p = pubsub
@@ -350,7 +389,7 @@ class TestPubSubMessages:
         assert await r.publish("foo", "test message") == 1
         assert await wait_for_message(p) is None
         assert self.message == make_message("message", "foo", "test message")
-        await p.close()
+        await p.aclose()
 
     async def test_channel_async_message_handler(self, r):
         p = r.pubsub(ignore_subscribe_messages=True)
@@ -359,7 +398,7 @@ class TestPubSubMessages:
         assert await r.publish("foo", "test message") == 1
         assert await wait_for_message(p) is None
         assert self.async_message == make_message("message", "foo", "test message")
-        await p.close()
+        await p.aclose()
 
     async def test_channel_sync_async_message_handler(self, r):
         p = r.pubsub(ignore_subscribe_messages=True)
@@ -371,7 +410,7 @@ class TestPubSubMessages:
         assert await wait_for_message(p) is None
         assert self.message == make_message("message", "foo", "test message")
         assert self.async_message == make_message("message", "bar", "test message 2")
-        await p.close()
+        await p.aclose()
 
     @pytest.mark.onlynoncluster
     async def test_pattern_message_handler(self, r: redis.Redis):
@@ -383,7 +422,7 @@ class TestPubSubMessages:
         assert self.message == make_message(
             "pmessage", "foo", "test message", pattern="f*"
         )
-        await p.close()
+        await p.aclose()
 
     async def test_unicode_channel_message_handler(self, r: redis.Redis):
         p = r.pubsub(ignore_subscribe_messages=True)
@@ -394,7 +433,7 @@ class TestPubSubMessages:
         assert await r.publish(channel, "test message") == 1
         assert await wait_for_message(p) is None
         assert self.message == make_message("message", channel, "test message")
-        await p.close()
+        await p.aclose()
 
     @pytest.mark.onlynoncluster
     # see: https://redis-py-cluster.readthedocs.io/en/stable/pubsub.html
@@ -410,7 +449,7 @@ class TestPubSubMessages:
         assert self.message == make_message(
             "pmessage", channel, "test message", pattern=pattern
         )
-        await p.close()
+        await p.aclose()
 
     async def test_get_message_without_subscribe(self, r: redis.Redis, pubsub):
         p = pubsub
@@ -524,7 +563,7 @@ class TestPubSubAutoDecoding:
         await r.publish(self.channel, new_data)
         assert await wait_for_message(p) is None
         assert self.message == self.make_message("message", self.channel, new_data)
-        await p.close()
+        await p.aclose()
 
     async def test_pattern_message_handler(self, r: redis.Redis):
         p = r.pubsub(ignore_subscribe_messages=True)
@@ -546,7 +585,7 @@ class TestPubSubAutoDecoding:
         assert self.message == self.make_message(
             "pmessage", self.channel, new_data, pattern=self.pattern
         )
-        await p.close()
+        await p.aclose()
 
     async def test_context_manager(self, r: redis.Redis):
         async with r.pubsub() as pubsub:
@@ -556,7 +595,7 @@ class TestPubSubAutoDecoding:
         assert pubsub.connection is None
         assert pubsub.channels == {}
         assert pubsub.patterns == {}
-        await pubsub.close()
+        await pubsub.aclose()
 
 
 @pytest.mark.onlynoncluster
@@ -597,9 +636,9 @@ class TestPubSubSubcommands:
 
         channels = [(b"foo", 1), (b"bar", 2), (b"baz", 3)]
         assert await r.pubsub_numsub("foo", "bar", "baz") == channels
-        await p1.close()
-        await p2.close()
-        await p3.close()
+        await p1.aclose()
+        await p2.aclose()
+        await p3.aclose()
 
     @skip_if_server_version_lt("2.8.0")
     async def test_pubsub_numpat(self, r: redis.Redis):
@@ -608,7 +647,7 @@ class TestPubSubSubcommands:
         for i in range(3):
             assert (await wait_for_message(p))["type"] == "psubscribe"
         assert await r.pubsub_numpat() == 3
-        await p.close()
+        await p.aclose()
 
 
 @pytest.mark.onlynoncluster
@@ -621,7 +660,7 @@ class TestPubSubPings:
         assert await wait_for_message(p) == make_message(
             type="pong", channel=None, data="", pattern=None
         )
-        await p.close()
+        await p.aclose()
 
     @skip_if_server_version_lt("3.0.0")
     async def test_send_pubsub_ping_message(self, r: redis.Redis):
@@ -631,7 +670,7 @@ class TestPubSubPings:
         assert await wait_for_message(p) == make_message(
             type="pong", channel=None, data="hello world", pattern=None
         )
-        await p.close()
+        await p.aclose()
 
 
 @pytest.mark.onlynoncluster
diff --git a/tests/test_asyncio/test_sentinel_managed_connection.py b/tests/test_asyncio/test_sentinel_managed_connection.py
index e784690c..711b3ee7 100644
--- a/tests/test_asyncio/test_sentinel_managed_connection.py
+++ b/tests/test_asyncio/test_sentinel_managed_connection.py
@@ -10,11 +10,11 @@ from .compat import mock
 pytestmark = pytest.mark.asyncio
 
 
-async def test_connect_retry_on_timeout_error():
+async def test_connect_retry_on_timeout_error(connect_args):
     """Test that the _connect function is retried in case of a timeout"""
     connection_pool = mock.AsyncMock()
     connection_pool.get_master_address = mock.AsyncMock(
-        return_value=("localhost", 6379)
+        return_value=(connect_args["host"], connect_args["port"])
     )
     conn = SentinelManagedConnection(
         retry_on_timeout=True,
diff --git a/tests/test_commands.py b/tests/test_commands.py
index 114fb6b6..b538dc30 100644
--- a/tests/test_commands.py
+++ b/tests/test_commands.py
@@ -4909,7 +4909,6 @@ class TestRedisCommands:
     def test_latency_reset(self, r: redis.Redis):
         assert r.latency_reset() == 0
 
-    @pytest.mark.onlynoncluster
     @skip_if_server_version_lt("4.0.0")
     @skip_if_redis_enterprise()
     def test_module_list(self, r):
diff --git a/tests/test_connection_pool.py b/tests/test_connection_pool.py
index ab0fc6be..ef70a8ff 100644
--- a/tests/test_connection_pool.py
+++ b/tests/test_connection_pool.py
@@ -1,6 +1,7 @@
 import os
 import re
 import time
+from contextlib import closing
 from threading import Thread
 from unittest import mock
 
@@ -51,6 +52,16 @@ class TestConnectionPool:
         assert isinstance(connection, DummyConnection)
         assert connection.kwargs == connection_kwargs
 
+    def test_closing(self):
+        connection_kwargs = {"foo": "bar", "biz": "baz"}
+        pool = redis.ConnectionPool(
+            connection_class=DummyConnection,
+            max_connections=None,
+            **connection_kwargs,
+        )
+        with closing(pool):
+            pass
+
     def test_multiple_connections(self, master_host):
         connection_kwargs = {"host": master_host[0], "port": master_host[1]}
         pool = self.get_pool(connection_kwargs=connection_kwargs)
diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py
index 7b048eec..e64a763b 100644
--- a/tests/test_pipeline.py
+++ b/tests/test_pipeline.py
@@ -1,3 +1,6 @@
+from contextlib import closing
+from unittest import mock
+
 import pytest
 import redis
 
@@ -284,6 +287,24 @@ class TestPipeline:
             assert unwatch_command is not None
             assert unwatch_command["command"] == "UNWATCH"
 
+    @pytest.mark.onlynoncluster
+    def test_close_is_reset(self, r):
+        with r.pipeline() as pipe:
+            called = 0
+
+            def mock_reset():
+                nonlocal called
+                called += 1
+
+            with mock.patch.object(pipe, "reset", mock_reset):
+                pipe.close()
+                assert called == 1
+
+    @pytest.mark.onlynoncluster
+    def test_closing(self, r):
+        with closing(r.pipeline()):
+            pass
+
     @pytest.mark.onlynoncluster
     def test_transaction_callable(self, r):
         r["a"] = 1