ResponseFuture: do not return the stream ID on client timeout

When a timeout occurs, the ResponseFuture associated with the query
returns its stream ID to the associated connection's free stream ID pool
- so that the stream ID can be immediately reused by another query.

However, that it incorrect and dangerous. If query A times out before it
receives a response from the cluster, a different query B might be
issued on the same connection and stream. If response for query A
arrives earlier than the response for query B, the first one might be
misinterpreted as the response for query B.

This commit changes the logic so that stream IDs are not returned on
timeout - now, they are only returned after receiving a response.

Author: piodul

cassandra/cluster.py

@@ -4361,8 +4361,11 @@ def _on_timeout(self, _attempts=0):
 
             pool = self.session._pools.get(self._current_host)
             if pool and not pool.is_shutdown:
-                with self._connection.lock:
-                    self._connection.request_ids.append(self._req_id)
+                # Do not return the stream ID to the pool yet. We cannot reuse it
+                # because the node might still be processing the query and will
+                # return a late response to that query - if we used such stream
+                # before the response to the previous query has arrived, the new
+                # query could get a response from the old query
 
                 pool.return_connection(self._connection)
 

cassandra/connection.py

@@ -1193,6 +1193,8 @@ def process_msg(self, header, body):
                 # This can only happen if the stream_id was
                 # removed due to an OperationTimedOut
                 except KeyError:
+                    with self.lock:
+                        self.request_ids.append(stream_id)
                     return
 
         try:

tests/unit/test_response_future.py

@@ -17,6 +17,8 @@
 except ImportError:
     import unittest # noqa
 
+from collections import deque
+from threading import RLock
 from mock import Mock, MagicMock, ANY
 
 from cassandra import ConsistencyLevel, Unavailable, SchemaTargetType, SchemaChangeType, OperationTimedOut
@@ -604,3 +606,28 @@ def test_repeat_orig_query_after_succesful_reprepare(self):
         rf._query = Mock(return_value=True)
         rf._execute_after_prepare('host', None, None, response)
         rf._query.assert_called_once_with('host')
+
+    def test_timeout_does_not_release_stream_id(self):
+        """
+        Make sure that stream ID is not reused immediately after client-side
+        timeout. Otherwise, a new request could reuse the stream ID and would
+        risk getting a response for the old, timed out query.
+        """
+        session = self.make_basic_session()
+        session.cluster._default_load_balancing_policy.make_query_plan.return_value = [Mock(endpoint='ip1'), Mock(endpoint='ip2')]
+        pool = self.make_pool()
+        session._pools.get.return_value = pool
+        connection = Mock(spec=Connection, lock=RLock(), _requests={}, request_ids=deque())
+        pool.borrow_connection.return_value = (connection, 1)
+
+        rf = self.make_response_future(session)
+        rf.send_request()
+
+        connection._requests[1] = (connection._handle_options_response, ProtocolHandler.decode_message, [])
+
+        rf._on_timeout()
+        pool.return_connection.assert_called_once_with(connection)
+        self.assertRaisesRegexp(OperationTimedOut, "Client request timeout", rf.result)
+
+        assert len(connection.request_ids) == 0, \
+            "Request IDs should be empty but it's not: {}".format(connection.request_ids)