Add timeout and in-flight observability to OperationTimedOut

Improve timeout observability in the driver, inspired by the Go driver
PR scylladb/gocql#847.

OperationTimedOut now carries optional timeout and in_flight fields that
are appended to the exception message when present (e.g.
"(timeout=10.0s, in_flight=42)"). All seven production raise sites in
connection.py and cluster.py pass these values where available.

Additionally, debug-level log lines are emitted for:
- Client-side request timeouts (host, timeout, in_flight, orphaned)
- Server-side read/write timeouts (host, consistency, received/required,
  data_retrieved/write_type, retry decision)

A helper _retry_decision_name() translates RetryPolicy constants to
human-readable strings for the log messages.

New keyword-only parameters are backward compatible — existing callers
that pass only positional errors/last_host continue to work unchanged.

Fixes: DRIVER-538

Signed-off-by: Yaniv Kaul <yaniv.kaul@scylladb.com>

Author: mykaul

cassandra/__init__.py

@@ -687,10 +687,29 @@ class OperationTimedOut(DriverException):
     The last :class:`~.Host` this operation was attempted against.
     """
 
-    def __init__(self, errors=None, last_host=None):
+    timeout = None
+    """
+    The timeout value (in seconds) that was in effect when the operation
+    timed out, or ``None`` if not applicable.
+    """
+
+    in_flight = None
+    """
+    The number of in-flight requests on the connection at the time of
+    the timeout (includes orphaned requests), or ``None`` if not applicable.
+    """
+
+    def __init__(self, errors=None, last_host=None, timeout=None, in_flight=None):
         self.errors = errors
         self.last_host = last_host
+        self.timeout = timeout
+        self.in_flight = in_flight
         message = "errors=%s, last_host=%s" % (self.errors, self.last_host)
+        if self.timeout is not None:
+            message += " (timeout=%ss" % self.timeout
+            if self.in_flight is not None:
+                message += ", in_flight=%d" % self.in_flight
+            message += ")"
         Exception.__init__(self, message)
 
 

cassandra/cluster.py

@@ -191,7 +191,6 @@ def _connection_reduce_fn(val,import_fn):
 
 log = logging.getLogger(__name__)
 
-
 _GRAPH_PAGING_MIN_DSE_VERSION = Version('6.8.0')
 
 _NOT_SET = object()
@@ -1683,7 +1682,8 @@ def add_execution_profile(self, name, profile, pool_wait_timeout=5):
             futures.update(session.update_created_pools())
         _, not_done = wait_futures(futures, pool_wait_timeout)
         if not_done:
-            raise OperationTimedOut("Failed to create all new connection pools in the %ss timeout." % pool_wait_timeout)
+            raise OperationTimedOut("Failed to create all new connection pools in the %ss timeout." % pool_wait_timeout,
+                                    timeout=pool_wait_timeout)
 
     def connection_factory(self, endpoint, host_conn = None, *args, **kwargs):
         """
@@ -4505,6 +4505,7 @@ def _on_timeout(self, _attempts=0):
             )
             return
 
+        conn_in_flight = None
         if self._connection is not None:
             try:
                 self._connection._requests.pop(self._req_id)
@@ -4515,9 +4516,14 @@ def _on_timeout(self, _attempts=0):
             except KeyError:
                 key = "Connection defunct by heartbeat"
                 errors = {key: "Client request timeout. See Session.execute[_async](timeout)"}
-                self._set_final_exception(OperationTimedOut(errors, self._current_host))
+                self._set_final_exception(OperationTimedOut(errors, self._current_host,
+                                                            timeout=self.timeout,
+                                                            in_flight=self._connection.in_flight))
                 return
 
+            # Capture connection stats before pool.return_connection() can alter state
+            conn_in_flight = self._connection.in_flight
+
             pool = self.session._pools.get(self._current_host)
             if pool and not pool.is_shutdown:
                 # Do not return the stream ID to the pool yet. We cannot reuse it
@@ -4542,7 +4548,9 @@ def _on_timeout(self, _attempts=0):
                 host = str(connection.endpoint) if connection else 'unknown'
                 errors = {host: "Request timed out while waiting for schema agreement. See Session.execute[_async](timeout) and Cluster.max_schema_agreement_wait."}
 
-        self._set_final_exception(OperationTimedOut(errors, self._current_host))
+        self._set_final_exception(OperationTimedOut(errors, self._current_host,
+                                                    timeout=self.timeout,
+                                                    in_flight=conn_in_flight))
 
     def _on_speculative_execute(self):
         self._timer = None

cassandra/connection.py

@@ -984,7 +984,8 @@ def factory(cls, endpoint, timeout, host_conn = None, *args, **kwargs):
             raise conn.last_error
         elif not conn.connected_event.is_set():
             conn.close()
-            raise OperationTimedOut("Timed out creating connection (%s seconds)" % timeout)
+            raise OperationTimedOut("Timed out creating connection (%s seconds)" % timeout,
+                                    timeout=timeout)
         else:
             return conn
 
@@ -1247,6 +1248,7 @@ def wait_for_responses(self, *msgs, **kwargs):
                 msg += ": %s" % (self.last_error,)
             raise ConnectionShutdown(msg)
         timeout = kwargs.get('timeout')
+        original_timeout = timeout  # preserve for exception reporting
         fail_on_error = kwargs.get('fail_on_error', True)
         waiter = ResponseWaiter(self, len(msgs), fail_on_error)
 
@@ -1271,7 +1273,8 @@ def wait_for_responses(self, *msgs, **kwargs):
                 if timeout is not None:
                     timeout -= 0.01
                     if timeout <= 0.0:
-                        raise OperationTimedOut()
+                        raise OperationTimedOut(timeout=original_timeout,
+                                                in_flight=self.in_flight)
                 time.sleep(0.01)
 
         try:
@@ -1796,7 +1799,8 @@ def deliver(self, timeout=None):
         if self.error:
             raise self.error
         elif not self.event.is_set():
-            raise OperationTimedOut()
+            raise OperationTimedOut(timeout=timeout,
+                                    in_flight=self.connection.in_flight)
         else:
             return self.responses
 
@@ -1823,7 +1827,10 @@ def wait(self, timeout):
             if self._exception:
                 raise self._exception
         else:
-            raise OperationTimedOut("Connection heartbeat timeout after %s seconds" % (timeout,), self.connection.endpoint)
+            raise OperationTimedOut("Connection heartbeat timeout after %s seconds" % (timeout,),
+                                    self.connection.endpoint,
+                                    timeout=timeout,
+                                    in_flight=self.connection.in_flight)
 
     def _options_callback(self, response):
         if isinstance(response, SupportedMessage):

tests/unit/test_cluster.py

@@ -87,6 +87,64 @@ def test_exception_types(self):
         assert issubclass(UnsupportedOperation, DriverException)
 
 
+class OperationTimedOutTest(unittest.TestCase):
+
+    def test_message_without_timeout(self):
+        """Default message format when no timeout info is provided."""
+        exc = OperationTimedOut(errors={'host1': 'some error'}, last_host='host1')
+        msg = str(exc)
+        assert "errors={'host1': 'some error'}" in msg
+        assert "last_host=host1" in msg
+        assert "timeout=" not in msg
+        assert "in_flight=" not in msg
+
+    def test_message_with_timeout_and_in_flight(self):
+        """Message includes timeout and in_flight when both are provided."""
+        exc = OperationTimedOut(errors={'host1': 'err'}, last_host='host1',
+                                timeout=10.0, in_flight=42)
+        msg = str(exc)
+        assert "(timeout=10.0s, in_flight=42)" in msg
+
+    def test_message_with_timeout_no_in_flight(self):
+        """Message includes timeout but not in_flight when only timeout is set."""
+        exc = OperationTimedOut(timeout=5.0)
+        msg = str(exc)
+        assert "(timeout=5.0s)" in msg
+        assert "in_flight=" not in msg
+
+    def test_message_no_args(self):
+        """No-argument form should not crash and should have clean message."""
+        exc = OperationTimedOut()
+        msg = str(exc)
+        assert "errors=None, last_host=None" in msg
+        assert "timeout=" not in msg
+
+    def test_attributes_accessible(self):
+        """New and existing attributes should be readable."""
+        exc = OperationTimedOut(errors={'h': 'e'}, last_host='h',
+                                timeout=10.0, in_flight=42)
+        assert exc.errors == {'h': 'e'}
+        assert exc.last_host == 'h'
+        assert exc.timeout == 10.0
+        assert exc.in_flight == 42
+
+    def test_attributes_default_none(self):
+        """New attributes should default to None when not provided."""
+        exc = OperationTimedOut()
+        assert exc.timeout is None
+        assert exc.in_flight is None
+        assert exc.errors is None
+        assert exc.last_host is None
+
+    def test_backward_compat_positional(self):
+        """Existing two-positional-arg form should still work."""
+        exc = OperationTimedOut({'h': 'err'}, 'host1')
+        assert exc.errors == {'h': 'err'}
+        assert exc.last_host == 'host1'
+        assert exc.timeout is None
+        assert exc.in_flight is None
+
+
 class ClusterTest(unittest.TestCase):
 
     def test_tuple_for_contact_points(self):

tests/unit/test_connection.py

@@ -520,6 +520,8 @@ def send_msg(msg, req_id, msg_callback):
         assert isinstance(exc, OperationTimedOut)
         assert exc.errors == 'Connection heartbeat timeout after 0.05 seconds'
         assert exc.last_host == DefaultEndPoint('localhost')
+        assert exc.timeout == 0.05
+        assert isinstance(exc.in_flight, int)
         holder.return_connection.assert_has_calls(
             [call(connection)] * get_holders.call_count)
 

tests/unit/test_response_future.py

@@ -142,6 +142,8 @@ def test_heartbeat_defunct_deadlock(self):
 
         connection = MagicMock(spec=Connection)
         connection._requests = {}
+        connection.in_flight = 5
+        connection.orphaned_request_ids = set()
 
         pool = Mock()
         pool.is_shutdown = False
@@ -162,8 +164,10 @@ def test_heartbeat_defunct_deadlock(self):
 
         # Simulate ResponseFuture timing out
         rf._on_timeout()
-        with pytest.raises(OperationTimedOut, match="Connection defunct by heartbeat"):
+        with pytest.raises(OperationTimedOut, match="Connection defunct by heartbeat") as exc_info:
             rf.result()
+        assert exc_info.value.timeout == 1
+        assert exc_info.value.in_flight == 5
 
     def test_read_timeout_error_message(self):
         session = self.make_session()
@@ -653,7 +657,7 @@ def test_timeout_does_not_release_stream_id(self):
         pool = self.make_pool()
         session._pools.get.return_value = pool
         connection = Mock(spec=Connection, lock=RLock(), _requests={}, request_ids=deque(),
-                orphaned_request_ids=set(), orphaned_threshold=256)
+                orphaned_request_ids=set(), orphaned_threshold=256, in_flight=3)
         pool.borrow_connection.return_value = (connection, 1)
 
         rf = self.make_response_future(session)
@@ -663,8 +667,10 @@ def test_timeout_does_not_release_stream_id(self):
 
         rf._on_timeout()
         pool.return_connection.assert_called_once_with(connection, stream_was_orphaned=True)
-        with pytest.raises(OperationTimedOut, match="Client request timeout"):
+        with pytest.raises(OperationTimedOut, match="Client request timeout") as exc_info:
             rf.result()
+        assert exc_info.value.timeout == 1
+        assert exc_info.value.in_flight == 3
 
         assert len(connection.request_ids) == 0, \
             "Request IDs should be empty but it's not: {}".format(connection.request_ids)