Repository: redis/redis-py
Pull request #3104 (2024-01-09 15:48:59)

Fix possible pipeline connections leak
### Pull Request check-list

_Please make sure to review and check all of these items:_

- [x] Do tests and lints pass with this change?
- [ ] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [x] 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
When Executing "n.write()" may generate some unknown errors(e.g. DataError), which could result in the connection not being released, finally lead to connection leak.

codecov-commenter: ## [Codecov](https://app.codecov.io/gh/redis/redis-py/pull/3104?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) Report
Attention: `7 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`18bb016`)](https://app.codecov.io/gh/redis/redis-py/commit/18bb01686e3af2f8ad086f838ce95672a78c12c4?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) 92.05% compared to head [(`58829d9`)](https://app.codecov.io/gh/redis/redis-py/pull/3104?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) 77.20%.

| [Files](https://app.codecov.io/gh/redis/redis-py/pull/3104?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | Patch % | Lines |
|---|---|---|
| [redis/cluster.py](https://app.codecov.io/gh/redis/redis-py/pull/3104?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY2x1c3Rlci5weQ==) | 0.00% | [7 Missing :warning: ](https://app.codecov.io/gh/redis/redis-py/pull/3104?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) |

:exclamation: Your organization needs to install the [Codecov GitHub app](https://github.com/apps/codecov/installations/select_target) to enable full functionality.

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


```diff
@@             Coverage Diff             @@
##           master    #3104       +/-   ##
===========================================
- Coverage   92.05%   77.20%   -14.86%     
===========================================
  Files         128      128               
  Lines       33195    33175       -20     
===========================================
- Hits        30557    25612     -4945     
- Misses       2638     7563     +4925     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/redis/redis-py/pull/3104?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).

dvora-h: @ING-XIAOJIAN thanks for suggestion, looks good, but WDYT about:
```
    try:
        node_commands = nodes.values()
        for n in node_commands:
            n.write()

        for n in node_commands:
            n.read()
    finally:
        for n in nodes.values():
            n.connection_pool.release(n.connection)
```

Patch:
diff --git a/redis/_cache.py b/redis/_cache.py
index 4255afb7..f8cc7d21 100644
--- a/redis/_cache.py
+++ b/redis/_cache.py
@@ -1,3 +1,4 @@
+import copy
 import random
 import time
 from collections import OrderedDict, defaultdict
@@ -226,7 +227,7 @@ class _LocalCache:
                 self.delete(command)
                 return
             self._update_access(command)
-            return self.cache[command]["response"]
+            return copy.deepcopy(self.cache[command]["response"])
 
     def delete(self, command: str):
         """
diff --git a/redis/cluster.py b/redis/cluster.py
index 7bdf4c19..e558be16 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -2166,32 +2166,34 @@ class ClusterPipeline(RedisCluster):
         # we dont' multiplex on the sockets as they come available,
         # but that shouldn't make too much difference.
         node_commands = nodes.values()
-        for n in node_commands:
-            n.write()
-
-        for n in node_commands:
-            n.read()
-
-        # release all of the redis connections we allocated earlier
-        # back into the connection pool.
-        # we used to do this step as part of a try/finally block,
-        # but it is really dangerous to
-        # release connections back into the pool if for some
-        # reason the socket has data still left in it
-        # from a previous operation. The write and
-        # read operations already have try/catch around them for
-        # all known types of errors including connection
-        # and socket level errors.
-        # So if we hit an exception, something really bad
-        # happened and putting any oF
-        # these connections back into the pool is a very bad idea.
-        # the socket might have unread buffer still sitting in it,
-        # and then the next time we read from it we pass the
-        # buffered result back from a previous command and
-        # every single request after to that connection will always get
-        # a mismatched result.
-        for n in nodes.values():
-            n.connection_pool.release(n.connection)
+        try:
+            node_commands = nodes.values()
+            for n in node_commands:
+                n.write()
+
+            for n in node_commands:
+                n.read()
+        finally:
+            # release all of the redis connections we allocated earlier
+            # back into the connection pool.
+            # we used to do this step as part of a try/finally block,
+            # but it is really dangerous to
+            # release connections back into the pool if for some
+            # reason the socket has data still left in it
+            # from a previous operation. The write and
+            # read operations already have try/catch around them for
+            # all known types of errors including connection
+            # and socket level errors.
+            # So if we hit an exception, something really bad
+            # happened and putting any oF
+            # these connections back into the pool is a very bad idea.
+            # the socket might have unread buffer still sitting in it,
+            # and then the next time we read from it we pass the
+            # buffered result back from a previous command and
+            # every single request after to that connection will always get
+            # a mismatched result.
+            for n in nodes.values():
+                n.connection_pool.release(n.connection)
 
         # if the response isn't an exception it is a
         # valid response from the node
diff --git a/redis/commands/core.py b/redis/commands/core.py
index 2840dc03..6d81d760 100644
--- a/redis/commands/core.py
+++ b/redis/commands/core.py
@@ -5017,14 +5017,16 @@ class HashCommands(CommandsProtocol):
         """
         if key is None and not mapping and not items:
             raise DataError("'hset' with no key value pairs")
-        items = items or []
+        pieces = []
+        if items:
+            pieces.extend(items)
         if key is not None:
-            items.extend((key, value))
+            pieces.extend((key, value))
         if mapping:
             for pair in mapping.items():
-                items.extend(pair)
+                pieces.extend(pair)
 
-        return self.execute_command("HSET", name, *items)
+        return self.execute_command("HSET", name, *pieces)
 
     def hsetnx(self, name: str, key: str, value: str) -> Union[Awaitable[bool], bool]:
         """
@@ -5260,12 +5262,12 @@ class ScriptCommands(CommandsProtocol):
     """
 
     def _eval(
-        self, command: str, script: str, numkeys: int, *keys_and_args: list
+        self, command: str, script: str, numkeys: int, *keys_and_args: str
     ) -> Union[Awaitable[str], str]:
         return self.execute_command(command, script, numkeys, *keys_and_args)
 
     def eval(
-        self, script: str, numkeys: int, *keys_and_args: list
+        self, script: str, numkeys: int, *keys_and_args: str
     ) -> Union[Awaitable[str], str]:
         """
         Execute the Lua ``script``, specifying the ``numkeys`` the script
@@ -5280,7 +5282,7 @@ class ScriptCommands(CommandsProtocol):
         return self._eval("EVAL", script, numkeys, *keys_and_args)
 
     def eval_ro(
-        self, script: str, numkeys: int, *keys_and_args: list
+        self, script: str, numkeys: int, *keys_and_args: str
     ) -> Union[Awaitable[str], str]:
         """
         The read-only variant of the EVAL command
@@ -5299,7 +5301,7 @@ class ScriptCommands(CommandsProtocol):
         return self.execute_command(command, sha, numkeys, *keys_and_args)
 
     def evalsha(
-        self, sha: str, numkeys: int, *keys_and_args: list
+        self, sha: str, numkeys: int, *keys_and_args: str
     ) -> Union[Awaitable[str], str]:
         """
         Use the ``sha`` to execute a Lua script already registered via EVAL
@@ -5315,7 +5317,7 @@ class ScriptCommands(CommandsProtocol):
         return self._evalsha("EVALSHA", sha, numkeys, *keys_and_args)
 
     def evalsha_ro(
-        self, sha: str, numkeys: int, *keys_and_args: list
+        self, sha: str, numkeys: int, *keys_and_args: str
     ) -> Union[Awaitable[str], str]:
         """
         The read-only variant of the EVALSHA command
diff --git a/redis/commands/json/commands.py b/redis/commands/json/commands.py
index 4c2e5836..b37dc376 100644
--- a/redis/commands/json/commands.py
+++ b/redis/commands/json/commands.py
@@ -121,7 +121,9 @@ class JSONCommands:
         """  # noqa
         return self.execute_command("JSON.OBJKEYS", name, str(path), keys=[name])
 
-    def objlen(self, name: str, path: Optional[str] = Path.root_path()) -> int:
+    def objlen(
+        self, name: str, path: Optional[str] = Path.root_path()
+    ) -> List[Optional[int]]:
         """Return the length of the dictionary JSON value under ``path`` at key
         ``name``.
 


Test patch:
diff --git a/tests/test_asyncio/test_cache.py b/tests/test_asyncio/test_cache.py
index 098ede8d..bf20337d 100644
--- a/tests/test_asyncio/test_cache.py
+++ b/tests/test_asyncio/test_cache.py
@@ -131,6 +131,17 @@ class TestLocalCache:
         assert cache.get(("LLEN", "mylist")) is None
         assert cache.get(("LINDEX", "mylist", 1)) == b"bar"
 
+    @pytest.mark.parametrize("r", [{"cache": _LocalCache()}], indirect=True)
+    async def test_cache_return_copy(self, r):
+        r, cache = r
+        await r.lpush("mylist", "foo", "bar", "baz")
+        assert await r.lrange("mylist", 0, -1) == [b"baz", b"bar", b"foo"]
+        res = cache.get(("LRANGE", "mylist", 0, -1))
+        assert res == [b"baz", b"bar", b"foo"]
+        res.append(b"new")
+        check = cache.get(("LRANGE", "mylist", 0, -1))
+        assert check == [b"baz", b"bar", b"foo"]
+
 
 @pytest.mark.skipif(HIREDIS_AVAILABLE, reason="PythonParser only")
 @pytest.mark.onlycluster
diff --git a/tests/test_cache.py b/tests/test_cache.py
index 570385a4..4eb5160e 100644
--- a/tests/test_cache.py
+++ b/tests/test_cache.py
@@ -15,6 +15,7 @@ def r(request):
         redis.Redis, request, protocol=3, client_cache=cache, **kwargs
     ) as client:
         yield client, cache
+        # client.flushdb()
 
 
 @pytest.mark.skipif(HIREDIS_AVAILABLE, reason="PythonParser only")
@@ -134,6 +135,17 @@ class TestLocalCache:
         assert cache.get(("LLEN", "mylist")) is None
         assert cache.get(("LINDEX", "mylist", 1)) == b"bar"
 
+    @pytest.mark.parametrize("r", [{"cache": _LocalCache()}], indirect=True)
+    def test_cache_return_copy(self, r):
+        r, cache = r
+        r.lpush("mylist", "foo", "bar", "baz")
+        assert r.lrange("mylist", 0, -1) == [b"baz", b"bar", b"foo"]
+        res = cache.get(("LRANGE", "mylist", 0, -1))
+        assert res == [b"baz", b"bar", b"foo"]
+        res.append(b"new")
+        check = cache.get(("LRANGE", "mylist", 0, -1))
+        assert check == [b"baz", b"bar", b"foo"]
+
 
 @pytest.mark.skipif(HIREDIS_AVAILABLE, reason="PythonParser only")
 @pytest.mark.onlycluster