Fix bug: client side caching causes unexpected disconnections (async version) (#3165)

* fix disconnects

* skip test in cluster

* add test

* save return value from handle_push_response (without it 'read_response' return the push message)

* insert return response from cache to the try block to prevent connection leak

* enable to get connection with data avaliable to read in csc mode and change can_read_destructive to not read data

* fix check if socket is empty (at_eof() can return False but this doesn't mean there's definitely more data to read)

---------

Co-authored-by: Chayim <chayim@users.noreply.github.com>

Author: dvora-h

redis/_parsers/base.py

@@ -182,7 +182,7 @@ async def can_read_destructive(self) -> bool:
             return True
         try:
             async with async_timeout(0):
-                return await self._stream.read(1)
+                return self._stream.at_eof()
         except TimeoutError:
             return False
 

redis/_parsers/resp3.py

@@ -261,7 +261,9 @@ async def _read_response(
                 )
                 for _ in range(int(response))
             ]
-            await self.handle_push_response(response, disable_decoding, push_request)
+            response = await self.handle_push_response(
+                response, disable_decoding, push_request
+            )
         else:
             raise InvalidResponse(f"Protocol Error: {raw!r}")
 

redis/asyncio/client.py

@@ -629,25 +629,27 @@ async def execute_command(self, *args, **options):
         pool = self.connection_pool
         conn = self.connection or await pool.get_connection(command_name, **options)
         response_from_cache = await conn._get_from_local_cache(args)
-        if response_from_cache is not None:
-            return response_from_cache
-        else:
-            if self.single_connection_client:
-                await self._single_conn_lock.acquire()
-            try:
-                response = await conn.retry.call_with_retry(
-                    lambda: self._send_command_parse_response(
-                        conn, command_name, *args, **options
-                    ),
-                    lambda error: self._disconnect_raise(conn, error),
-                )
-                conn._add_to_local_cache(args, response, keys)
-                return response
-            finally:
-                if self.single_connection_client:
-                    self._single_conn_lock.release()
-                if not self.connection:
-                    await pool.release(conn)
+        try:
+            if response_from_cache is not None:
+                return response_from_cache
+            else:
+                try:
+                    if self.single_connection_client:
+                        await self._single_conn_lock.acquire()
+                    response = await conn.retry.call_with_retry(
+                        lambda: self._send_command_parse_response(
+                            conn, command_name, *args, **options
+                        ),
+                        lambda error: self._disconnect_raise(conn, error),
+                    )
+                    conn._add_to_local_cache(args, response, keys)
+                    return response
+                finally:
+                    if self.single_connection_client:
+                        self._single_conn_lock.release()
+        finally:
+            if not self.connection:
+                await pool.release(conn)
 
     async def parse_response(
         self, connection: Connection, command_name: Union[str, bytes], **options

redis/asyncio/connection.py

@@ -685,7 +685,7 @@ def pack_commands(self, commands: Iterable[Iterable[EncodableT]]) -> List[bytes]
 
     def _socket_is_empty(self):
         """Check if the socket is empty"""
-        return not self._reader.at_eof()
+        return len(self._reader._buffer) == 0
 
     def _cache_invalidation_process(
         self, data: List[Union[str, Optional[List[str]]]]
@@ -1192,12 +1192,18 @@ def make_connection(self):
     async def ensure_connection(self, connection: AbstractConnection):
         """Ensure that the connection object is connected and valid"""
         await connection.connect()
-        # connections that the pool provides should be ready to send
-        # a command. if not, the connection was either returned to the
+        # if client caching is not enabled connections that the pool
+        # provides should be ready to send a command.
+        # if not, the connection was either returned to the
         # pool before all data has been read or the socket has been
         # closed. either way, reconnect and verify everything is good.
+        # (if caching enabled the connection will not always be ready
+        # to send a command because it may contain invalidation messages)
         try:
-            if await connection.can_read_destructive():
+            if (
+                await connection.can_read_destructive()
+                and connection.client_cache is None
+            ):
                 raise ConnectionError("Connection has data") from None
         except (ConnectionError, OSError):
             await connection.disconnect()

tests/test_asyncio/test_cache.py

@@ -142,6 +142,48 @@ async def test_cache_return_copy(self, r):
         check = cache.get(("LRANGE", "mylist", 0, -1))
         assert check == [b"baz", b"bar", b"foo"]
 
+    @pytest.mark.onlynoncluster
+    @pytest.mark.parametrize(
+        "r",
+        [{"cache": _LocalCache(), "kwargs": {"decode_responses": True}}],
+        indirect=True,
+    )
+    async def test_csc_not_cause_disconnects(self, r):
+        r, cache = r
+        id1 = await r.client_id()
+        await r.mset({"a": 1, "b": 1, "c": 1, "d": 1, "e": 1})
+        assert await r.mget("a", "b", "c", "d", "e") == ["1", "1", "1", "1", "1"]
+        id2 = await r.client_id()
+
+        # client should get value from client cache
+        assert await r.mget("a", "b", "c", "d", "e") == ["1", "1", "1", "1", "1"]
+        assert cache.get(("MGET", "a", "b", "c", "d", "e")) == [
+            "1",
+            "1",
+            "1",
+            "1",
+            "1",
+        ]
+
+        await r.mset({"a": 2, "b": 2, "c": 2, "d": 2, "e": 2})
+        id3 = await r.client_id()
+        # client should get value from redis server post invalidate messages
+        assert await r.mget("a", "b", "c", "d", "e") == ["2", "2", "2", "2", "2"]
+
+        await r.mset({"a": 3, "b": 3, "c": 3, "d": 3, "e": 3})
+        # need to check that we get correct value 3 and not 2
+        assert await r.mget("a", "b", "c", "d", "e") == ["3", "3", "3", "3", "3"]
+        # client should get value from client cache
+        assert await r.mget("a", "b", "c", "d", "e") == ["3", "3", "3", "3", "3"]
+
+        await r.mset({"a": 4, "b": 4, "c": 4, "d": 4, "e": 4})
+        # need to check that we get correct value 4 and not 3
+        assert await r.mget("a", "b", "c", "d", "e") == ["4", "4", "4", "4", "4"]
+        # client should get value from client cache
+        assert await r.mget("a", "b", "c", "d", "e") == ["4", "4", "4", "4", "4"]
+        id4 = await r.client_id()
+        assert id1 == id2 == id3 == id4
+
 
 @pytest.mark.skipif(HIREDIS_AVAILABLE, reason="PythonParser only")
 @pytest.mark.onlycluster