Repository: redis/redis-py
Pull request #2104 (2022-04-13 11:39:44)

Catch `Exception` and not `BaseException` in the `Connection`
### Pull Request check-list

_Please make sure to review and check all of these items:_

- [x] Does `$ tox` pass with this change (including linting)?
- [x] 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)?
- [x] Was the change added to CHANGES file?

### Description of change

Fixes Issue #2103:
Raising a BaseException inside a `Connection`'s send or receive oprations causes a `disconnect`().  BaseExceptions are commonly used for _out of band_ messages in Python and normally not be expecially acted upon, only be handled at the top-level 

See also:  #2356 which focuses on a different aspcect of error handling.

codecov-commenter: # [Codecov](https://codecov.io/gh/redis/redis-py/pull/2104?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) Report
> Merging [#2104](https://codecov.io/gh/redis/redis-py/pull/2104?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) (d4ad2d6) into [master](https://codecov.io/gh/redis/redis-py/commit/805b1842dbf7a732bad484e418dd7da4ec6ab890?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) (805b184) will **not change** coverage.
> The diff coverage is `50.00%`.

```diff
@@           Coverage Diff           @@
##           master    #2104   +/-   ##
=======================================
  Coverage   92.43%   92.43%           
=======================================
  Files         104      104           
  Lines       24363    24363           
=======================================
  Hits        22520    22520           
  Misses       1843     1843           
```


| [Impacted Files](https://codecov.io/gh/redis/redis-py/pull/2104?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | Coverage Δ | |
|---|---|---|
| [redis/asyncio/connection.py](https://codecov.io/gh/redis/redis-py/pull/2104/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9jb25uZWN0aW9uLnB5) | `84.49% <50.00%> (ø)` | |
| [redis/connection.py](https://codecov.io/gh/redis/redis-py/pull/2104/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY29ubmVjdGlvbi5weQ==) | `86.53% <50.00%> (ø)` | |
| [tests/test\_cluster.py](https://codecov.io/gh/redis/redis-py/pull/2104/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9jbHVzdGVyLnB5) | `97.48% <0.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/redis/redis-py/pull/2104?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/redis/redis-py/pull/2104?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis). Last update [805b184...d4ad2d6](https://codecov.io/gh/redis/redis-py/pull/2104?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).

chayim: @kristjanvalur Why did you catch Exception rather that BaseException? Was there an inheritance issue you saw?
WisdomPill: @chayim have a look here #2103
kristjanvalur: I can add tests.  As I explained in the defect, `BaseException` is usually not caught except in top level handlers, unless it is to implement something like `async_timeout.Timeout()`  .  `BaseException` was added to Python precisely to be a sort of _meta_ exception that regular exception handling doesn't touch, the `Exception` being the most general class usually touched in code that isn't trying to be too clever.
kristjanvalur: In fact, I don't see why there is a `except Exception: self.disconnect(); raise` at all. Doesn't seem like something the Connection should do if it gets an _unexpected_ exception.  IMHO, this clause should be removed alltogether.
kristjanvalur: Hm, now it is mysteriously failing.  it asserts on the "assert  pubsub.connection.is_connected"
I´m running tests both on windows and in a linux dev container using
`pytest tests/test_async/test_pubsub.py`.   I don't get the test failures above.
kristjanvalur: Well, it seems I cracked the unittest problem I was having.  Turned out that only in Python 3.8 did asyncio.CancelledError get promoted to BaseException.
Fixed up the test to skip older python versions and added Mock-based tests for exception propagation for both async and blocking redis.
kristjanvalur: any responses?  @WisdomPill ?
kristjanvalur: I'll clean up the commit history a bit...
kristjanvalur: yet another fluke test failure.
kristjanvalur: I added the Connection.disconnect(nowait) optional arg.  When `true`, the disconnect() call will not attempt to wait for the flushing of the write buffer during close().  This is an async operation, and when calling "disconnect" as a response to an error, the async close can cause other errors to be raised.  Notably, if running within a "timeout" context, a `CancelledError` may get thrown.

In the case of an error, there should be no need to wait for a flush of the write buffer.
akx: I bumped into this too – trying to e.g. ctrl+c out of py.test requires a couple of tries because `KeyboardInterrupt` (a `BaseException`) is caught where it really must not be.


Patch:
diff --git a/docs/examples/search_vector_similarity_examples.ipynb b/docs/examples/search_vector_similarity_examples.ipynb
index 42c03f0c..2b026109 100644
--- a/docs/examples/search_vector_similarity_examples.ipynb
+++ b/docs/examples/search_vector_similarity_examples.ipynb
@@ -74,7 +74,7 @@
     "r.hset(\"b\", \"v\", \"aaaabaaa\")\n",
     "r.hset(\"c\", \"v\", \"aaaaabaa\")\n",
     "\n",
-    "q = Query(\"*=>[KNN 2 @v $vec]\").return_field(\"__v_score\")\n",
+    "q = Query(\"*=>[KNN 2 @v $vec]\").return_field(\"__v_score\").dialect(2)\n",
     "r.ft().search(q, query_params={\"vec\": \"aaaaaaaa\"})"
    ]
   }
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index 16f33e2a..a470b6f7 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -502,8 +502,6 @@ class HiredisParser(BaseParser):
             # data was read from the socket and added to the buffer.
             # return True to indicate that data was read.
             return True
-        except asyncio.CancelledError:
-            raise
         except (socket.timeout, asyncio.TimeoutError):
             if raise_on_timeout:
                 raise TimeoutError("Timeout reading from socket") from None
@@ -721,7 +719,7 @@ class Connection:
                 lambda: self._connect(), lambda error: self.disconnect()
             )
         except asyncio.CancelledError:
-            raise
+            raise  # in 3.7 and earlier, this is an Exception, not BaseException
         except (socket.timeout, asyncio.TimeoutError):
             raise TimeoutError("Timeout connecting to server")
         except OSError as e:
@@ -916,7 +914,7 @@ class Connection:
             raise ConnectionError(
                 f"Error {err_no} while writing to socket. {errmsg}."
             ) from e
-        except BaseException:
+        except Exception:
             await self.disconnect()
             raise
 
@@ -958,7 +956,7 @@ class Connection:
             raise ConnectionError(
                 f"Error while reading from {self.host}:{self.port} : {e.args}"
             )
-        except BaseException:
+        except Exception:
             await self.disconnect()
             raise
 
diff --git a/redis/commands/search/aggregation.py b/redis/commands/search/aggregation.py
index 061e69c2..a171fa13 100644
--- a/redis/commands/search/aggregation.py
+++ b/redis/commands/search/aggregation.py
@@ -104,7 +104,6 @@ class AggregateRequest:
         self._aggregateplan = []
         self._loadfields = []
         self._loadall = False
-        self._limit = Limit()
         self._max = 0
         self._with_schema = False
         self._verbatim = False
@@ -211,7 +210,8 @@ class AggregateRequest:
         `sort_by()` instead.
 
         """
-        self._limit = Limit(offset, num)
+        _limit = Limit(offset, num)
+        self._aggregateplan.extend(_limit.build_args())
         return self
 
     def sort_by(self, *fields, **kwargs):
@@ -323,8 +323,6 @@ class AggregateRequest:
 
         ret.extend(self._aggregateplan)
 
-        ret += self._limit.build_args()
-
         return ret
 
 
diff --git a/redis/connection.py b/redis/connection.py
index 491df8e0..2e33e31d 100755
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -766,7 +766,7 @@ class Connection:
                 errno = e.args[0]
                 errmsg = e.args[1]
             raise ConnectionError(f"Error {errno} while writing to socket. {errmsg}.")
-        except BaseException:
+        except Exception:
             self.disconnect()
             raise
 
@@ -804,7 +804,7 @@ class Connection:
         except OSError as e:
             self.disconnect()
             raise ConnectionError(f"Error while reading from {hosterr}" f" : {e.args}")
-        except BaseException:
+        except Exception:
             self.disconnect()
             raise
 


Test patch:
diff --git a/tests/test_asyncio/test_pubsub.py b/tests/test_asyncio/test_pubsub.py
index 2ff0c726..555cfdbc 100644
--- a/tests/test_asyncio/test_pubsub.py
+++ b/tests/test_asyncio/test_pubsub.py
@@ -1,7 +1,9 @@
 import asyncio
 import functools
 import socket
+import sys
 from typing import Optional
+from unittest.mock import patch
 
 import async_timeout
 import pytest
@@ -914,3 +916,75 @@ class TestPubSubAutoReconnect:
                     return True
         except asyncio.TimeoutError:
             return False
+
+
+@pytest.mark.onlynoncluster
+class TestBaseException:
+    @pytest.mark.skipif(
+        sys.version_info < (3, 8), reason="requires python 3.8 or higher"
+    )
+    async def test_outer_timeout(self, r: redis.Redis):
+        """
+        Using asyncio_timeout manually outside the inner method timeouts works.
+        This works on Python versions 3.8 and greater, at which time asyncio.
+        CancelledError became a BaseException instead of an Exception before.
+        """
+        pubsub = r.pubsub()
+        await pubsub.subscribe("foo")
+        assert pubsub.connection.is_connected
+
+        async def get_msg_or_timeout(timeout=0.1):
+            async with async_timeout.timeout(timeout):
+                # blocking method to return messages
+                while True:
+                    response = await pubsub.parse_response(block=True)
+                    message = await pubsub.handle_message(
+                        response, ignore_subscribe_messages=False
+                    )
+                    if message is not None:
+                        return message
+
+        # get subscribe message
+        msg = await get_msg_or_timeout(10)
+        assert msg is not None
+        # timeout waiting for another message which never arrives
+        assert pubsub.connection.is_connected
+        with pytest.raises(asyncio.TimeoutError):
+            await get_msg_or_timeout()
+        # the timeout on the read should not cause disconnect
+        assert pubsub.connection.is_connected
+
+    async def test_base_exception(self, r: redis.Redis):
+        """
+        Manually trigger a BaseException inside the parser's .read_response method
+        and verify that it isn't caught
+        """
+        pubsub = r.pubsub()
+        await pubsub.subscribe("foo")
+        assert pubsub.connection.is_connected
+
+        async def get_msg():
+            # blocking method to return messages
+            while True:
+                response = await pubsub.parse_response(block=True)
+                message = await pubsub.handle_message(
+                    response, ignore_subscribe_messages=False
+                )
+                if message is not None:
+                    return message
+
+        # get subscribe message
+        msg = await get_msg()
+        assert msg is not None
+        # timeout waiting for another message which never arrives
+        assert pubsub.connection.is_connected
+        with patch("redis.asyncio.connection.PythonParser.read_response") as mock1:
+            mock1.side_effect = BaseException("boom")
+            with patch("redis.asyncio.connection.HiredisParser.read_response") as mock2:
+                mock2.side_effect = BaseException("boom")
+
+                with pytest.raises(BaseException):
+                    await get_msg()
+
+        # the timeout on the read should not cause disconnect
+        assert pubsub.connection.is_connected
diff --git a/tests/test_pubsub.py b/tests/test_pubsub.py
index 26df598b..5d86934d 100644
--- a/tests/test_pubsub.py
+++ b/tests/test_pubsub.py
@@ -735,3 +735,45 @@ class TestPubSubAutoReconnect:
         for message in self.pubsub.listen():
             self.messages.put(message)
             return True
+
+
+@pytest.mark.onlynoncluster
+class TestBaseException:
+    def test_base_exception(self, r: redis.Redis):
+        """
+        Manually trigger a BaseException inside the parser's .read_response method
+        and verify that it isn't caught
+        """
+        pubsub = r.pubsub()
+        pubsub.subscribe("foo")
+
+        def is_connected():
+            return pubsub.connection._sock is not None
+
+        assert is_connected()
+
+        def get_msg():
+            # blocking method to return messages
+            while True:
+                response = pubsub.parse_response(block=True)
+                message = pubsub.handle_message(
+                    response, ignore_subscribe_messages=False
+                )
+                if message is not None:
+                    return message
+
+        # get subscribe message
+        msg = get_msg()
+        assert msg is not None
+        # timeout waiting for another message which never arrives
+        assert is_connected()
+        with patch("redis.connection.PythonParser.read_response") as mock1:
+            mock1.side_effect = BaseException("boom")
+            with patch("redis.connection.HiredisParser.read_response") as mock2:
+                mock2.side_effect = BaseException("boom")
+
+                with pytest.raises(BaseException):
+                    get_msg()
+
+        # the timeout on the read should not cause disconnect
+        assert is_connected()