PYTHON-1024: fix error handling for SSL errors

This also removes a check that depends internal behavior that no longer
holds, as `process_io_buffer` is no longer called after `handle_read`
gets an error.

Author: mambocab

CHANGELOG.rst

@@ -1,6 +1,10 @@
 3.15.2
 ======
 
+Bug Fixes
+---------
+* Improve and fix socket error-catching code in nonblocking-socket reactors (PYTHON-1024)
+
 Other
 -----
 * Fix tests when RF is not maintained if we decomission a node (PYTHON-1017)

cassandra/io/asyncorereactor.py

@@ -424,10 +424,14 @@ def handle_read(self):
                     break
         except socket.error as err:
             if ssl and isinstance(err, ssl.SSLError):
-                if err.args[0] not in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE):
+                if err.args[0] in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE):
+                    return
+                else:
                     self.defunct(err)
                     return
-            elif err.args[0] not in NONBLOCKING:
+            elif err.args[0] in NONBLOCKING:
+                return
+            else:
                 self.defunct(err)
                 return
 

cassandra/io/libevreactor.py

@@ -344,10 +344,14 @@ def handle_read(self, watcher, revents, errno=None):
                     break
         except socket.error as err:
             if ssl and isinstance(err, ssl.SSLError):
-                if err.args[0] not in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE):
+                if err.args[0] in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE):
+                    return
+                else:
                     self.defunct(err)
                     return
-            elif err.args[0] not in NONBLOCKING:
+            elif err.args[0] in NONBLOCKING:
+                return
+            else:
                 self.defunct(err)
                 return
 

tests/unit/io/utils.py

@@ -284,7 +284,6 @@ def side_effect(*args):
 
         self.get_socket(c).recv.side_effect = side_effect
         c.handle_read(*self.null_handle_function_args)
-        self.assertEqual(c._current_frame.end_pos, 20000 + len(header))
         # the EAGAIN prevents it from reading the last 100 bytes
         c._iobuf.seek(0, os.SEEK_END)
         pos = c._iobuf.tell()