From 1cefc74d0fa855612aa0a1659028c7690d00a2d7 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Mon, 21 Sep 2026 21:50:16 -0400 Subject: [PATCH 1/4] control-connection: reconnect when DOWN handling is skipped ControlConnection._signal_error() treated a conviction-policy decision as proof that Cluster.on_down() had dispatched its asynchronous callbacks. DOWN handling can instead be suppressed by an open session pool, existing host state, disabled pool creation, shutdown, or a rejected executor submission, leaving the defunct control connection without recovery work. Report whether DOWN handling was actually queued and reconnect directly when it was not. Keep callback-driven recovery for accepted transitions so the normal host-down path does not submit a duplicate attempt, and cover TCP, Unix-socket, alternate-route, and stable host-ID connections. Fixes #847 --- cassandra/cluster.py | 77 ++++------ tests/unit/test_cluster.py | 178 ++++++++++++++++++++++++ tests/unit/test_control_connection.py | 193 ++++++++++++++++++++++++-- 3 files changed, 386 insertions(+), 62 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index a87b2d2082..3b2bec1ce4 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -184,18 +184,24 @@ def _future_completed(future): def run_in_executor(f): """ A decorator to run the given method in the ThreadPoolExecutor. + + The wrapper returns whether the call was handed to the executor. It is + False when the cluster is shutting down or the executor rejected the + submission, in which case the wrapped method never runs. """ @wraps(f) def new_f(self, *args, **kwargs): if self.is_shutdown: - return + return False try: future = self.executor.submit(f, self, *args, **kwargs) future.add_done_callback(_future_completed) + return True except Exception: log.exception("Failed to submit task to executor") + return False return new_f @@ -2025,9 +2031,11 @@ def on_down_potentially_blocking(self, host, is_host_addition): def on_down(self, host, is_host_addition, expect_host_to_be_down=False): """ Intended for internal use only. + + Returns whether DOWN handling was dispatched. """ if self.is_shutdown or self.allow_control_connection_query_fallback == ControlConnectionQueryFallback.SkipPoolCreation: - return + return False with host.lock: was_up = host.is_up @@ -2042,14 +2050,17 @@ def on_down(self, host, is_host_addition, expect_host_to_be_down=False): if pool_state: connected |= pool_state['open_count'] > 0 if connected: - return + return False host.set_down() if (not was_up and not expect_host_to_be_down) or host.is_currently_reconnecting(): - return + return False log.warning("Host %s has been marked down", host) - self.on_down_potentially_blocking(host, is_host_addition) + # on_down_potentially_blocking() reports whether the executor accepted + # it. A dropped submission runs no DOWN callback, so it must not be + # reported as dispatched. + return self.on_down_potentially_blocking(host, is_host_addition) def on_add(self, host, refresh_nodes=True): if self.is_shutdown: @@ -2139,10 +2150,11 @@ def on_remove(self, host): reconnection_handler.cancel() def signal_connection_failure(self, host, connection_exc, is_host_addition, expect_host_to_be_down=False): + """Return whether the failure caused DOWN handling to be dispatched.""" is_down = host.signal_connection_failure(connection_exc) - if is_down: - self.on_down(host, is_host_addition, expect_host_to_be_down) - return is_down + if not is_down: + return False + return self.on_down(host, is_host_addition, expect_host_to_be_down) def add_host(self, endpoint, datacenter=None, rack=None, signal=True, refresh_nodes=True, host_id=None): """ @@ -4585,51 +4597,18 @@ def _signal_error(self): if self._is_shutdown: return - # try just signaling the cluster, as this will trigger a reconnect - # as part of marking the host down + # If DOWN handling is dispatched, its control connection callback + # will reconnect. Otherwise reconnect directly. if self._connection and self._connection.is_defunct: connection = self._connection host = self._get_host_for_connection(connection) - # host may be None if it's already been removed, but that indicates - # that errors have already been reported, so we're fine - if host: - original_endpoint = getattr( - connection, 'original_endpoint', None) - unix_backed = ( - isinstance(host.endpoint, UnixSocketEndPoint) or - isinstance(connection.endpoint, UnixSocketEndPoint) or - isinstance(original_endpoint, UnixSocketEndPoint)) - route_mismatch = connection.endpoint != host.endpoint - # Keep ordinary endpoint-equal TCP connections on the - # legacy signal-only path. General suppressed-DOWN recovery - # and its reconnection cadence are outside this change. - if not unix_backed and not route_mismatch: - self._cluster.signal_connection_failure( - host, connection.last_error, - is_host_addition=False) - return - - # A newly resolvable Unix Host or alternate connection - # route still needs the direct reconnect fallback when - # host-state handling suppresses its DOWN notification. A - # fresh DOWN transition guarantees that on_down() will - # enqueue the reconnect instead. - with host.lock: - host_was_up = host.is_up is True - host_was_reconnecting = ( - host.is_currently_reconnecting()) - self._cluster.signal_connection_failure( - host, connection.last_error, - is_host_addition=False) - down_notification_queued = ( - host_was_up and not host_was_reconnecting and - host.is_up is False) - - if down_notification_queued: - return + if host and self._cluster.signal_connection_failure( + host, connection.last_error, + is_host_addition=False): + return - # if the connection is not defunct or the host already left, reconnect - # manually + # If the connection is not defunct, the host is unresolved, or DOWN + # handling was suppressed, reconnect manually. self.reconnect() def on_up(self, host): diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 74ed346c68..4fd6f30437 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -771,6 +771,184 @@ def test_set_keyspace_for_all_pools_reports_all_errors(self, *_): callback.assert_called_once() assert callback.call_args.args[0] == {'host1': [keyspace_error]} + +class ClusterDownHandlingTest(unittest.TestCase): + + def setUp(self): + self.cluster = Cluster(contact_points=[]) + self.addCleanup(self.cluster.shutdown) + self.cluster.profile_manager.distance = Mock( + return_value=HostDistance.LOCAL) + # The real method reports whether the executor accepted the work. + self.cluster.on_down_potentially_blocking = Mock(return_value=True) + self.host = Host( + "127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4()) + self.host.set_up() + + def test_signal_connection_failure_rejected_by_conviction_policy(self): + error = ConnectionException("connection failed") + self.host.signal_connection_failure = Mock(return_value=False) + self.cluster.on_down = Mock(return_value=True) + + assert not self.cluster.signal_connection_failure( + self.host, error, is_host_addition=False) + + self.host.signal_connection_failure.assert_called_once_with(error) + self.cluster.on_down.assert_not_called() + + def test_signal_connection_failure_returns_down_handling_result(self): + error = ConnectionException("connection failed") + self.host.signal_connection_failure = Mock(return_value=True) + self.cluster.on_down = Mock(return_value=False) + + assert not self.cluster.signal_connection_failure( + self.host, error, is_host_addition=True, + expect_host_to_be_down=True) + + self.cluster.on_down.assert_called_once_with( + self.host, True, True) + + def test_on_down_dispatches_normal_up_to_down_transition(self): + assert self.cluster.on_down(self.host, is_host_addition=False) + + assert self.host.is_up is False + self.cluster.on_down_potentially_blocking.assert_called_once_with( + self.host, False) + + def test_on_down_discounted_when_any_session_has_an_open_pool(self): + missing_pool = Mock() + missing_pool.get_pool_state.return_value = {} + closed_pool = Mock() + closed_pool.get_pool_state.return_value = { + self.host: {'open_count': 0}} + open_pool = Mock() + open_pool.get_pool_state.return_value = { + self.host: {'open_count': 1}} + self.cluster.sessions = [missing_pool, closed_pool, open_pool] + + assert not self.cluster.on_down(self.host, is_host_addition=False) + + assert self.host.is_up is True + self.cluster.on_down_potentially_blocking.assert_not_called() + + def test_on_down_not_discounted_when_all_pools_are_missing_or_closed(self): + missing_pool = Mock() + missing_pool.get_pool_state.return_value = {} + closed_pool = Mock() + closed_pool.get_pool_state.return_value = { + self.host: {'open_count': 0}} + self.cluster.sessions = [missing_pool, closed_pool] + + assert self.cluster.on_down(self.host, is_host_addition=False) + + self.cluster.on_down_potentially_blocking.assert_called_once_with( + self.host, False) + + def test_on_down_open_pool_discount_can_be_disabled(self): + open_pool = Mock() + open_pool.get_pool_state.return_value = { + self.host: {'open_count': 1}} + self.cluster.sessions = [open_pool] + self.cluster._discount_down_events = False + + assert self.cluster.on_down(self.host, is_host_addition=False) + + self.cluster.on_down_potentially_blocking.assert_called_once_with( + self.host, False) + + def test_on_down_does_not_discount_ignored_host(self): + open_pool = Mock() + open_pool.get_pool_state.return_value = { + self.host: {'open_count': 1}} + self.cluster.sessions = [open_pool] + self.cluster.profile_manager.distance.return_value = \ + HostDistance.IGNORED + + assert self.cluster.on_down(self.host, is_host_addition=False) + + self.cluster.on_down_potentially_blocking.assert_called_once_with( + self.host, False) + + def test_on_down_skipped_during_shutdown(self): + self.cluster.is_shutdown = True + + try: + assert not self.cluster.on_down( + self.host, is_host_addition=False) + finally: + # Keep the real shutdown cleanup effective for this test cluster. + self.cluster.is_shutdown = False + + assert self.host.is_up is True + self.cluster.on_down_potentially_blocking.assert_not_called() + + def test_on_down_skipped_when_pool_creation_is_disabled(self): + self.cluster.allow_control_connection_query_fallback = \ + ControlConnectionQueryFallback.SkipPoolCreation + + assert not self.cluster.on_down(self.host, is_host_addition=False) + + assert self.host.is_up is True + self.cluster.on_down_potentially_blocking.assert_not_called() + + def test_on_down_skips_already_down_or_uninitialized_host(self): + for initial_state in (False, None): + with self.subTest(initial_state=initial_state): + self.host.is_up = initial_state + self.cluster.on_down_potentially_blocking.reset_mock() + + assert not self.cluster.on_down( + self.host, is_host_addition=False) + + assert self.host.is_up is False + self.cluster.on_down_potentially_blocking.assert_not_called() + + def test_on_down_expected_down_host_dispatches_recovery(self): + for initial_state in (False, None): + with self.subTest(initial_state=initial_state): + self.host.is_up = initial_state + self.cluster.on_down_potentially_blocking.reset_mock() + + assert self.cluster.on_down( + self.host, is_host_addition=True, + expect_host_to_be_down=True) + + self.cluster.on_down_potentially_blocking \ + .assert_called_once_with(self.host, True) + + def test_on_down_existing_reconnector_prevents_duplicate_recovery(self): + self.host.get_and_set_reconnection_handler(Mock()) + + assert not self.cluster.on_down(self.host, is_host_addition=False) + + assert self.host.is_up is False + self.cluster.on_down_potentially_blocking.assert_not_called() + + def test_on_down_reports_undispatched_when_the_executor_rejects_it(self): + # Restore the real method so the executor submission is exercised. + del self.cluster.on_down_potentially_blocking + real_executor = self.cluster.executor + self.addCleanup(setattr, self.cluster, 'executor', real_executor) + self.cluster.executor = Mock() + self.cluster.executor.submit.side_effect = RuntimeError( + "cannot schedule new futures") + + assert not self.cluster.on_down(self.host, is_host_addition=False) + + assert self.host.is_up is False + self.cluster.executor.submit.assert_called_once() + + def test_on_down_reports_dispatched_when_the_executor_accepts_it(self): + del self.cluster.on_down_potentially_blocking + real_executor = self.cluster.executor + self.addCleanup(setattr, self.cluster, 'executor', real_executor) + self.cluster.executor = Mock() + + assert self.cluster.on_down(self.host, is_host_addition=False) + + self.cluster.executor.submit.assert_called_once() + + class ProtocolVersionTests(unittest.TestCase): def test_protocol_downgrade_test(self): diff --git a/tests/unit/test_control_connection.py b/tests/unit/test_control_connection.py index 13a68c7e55..b4a9224360 100644 --- a/tests/unit/test_control_connection.py +++ b/tests/unit/test_control_connection.py @@ -21,12 +21,14 @@ from cassandra.protocol import ResultMessage, RESULT_KIND_ROWS from cassandra.cluster import (Cluster, ControlConnection, _Scheduler, ProfileManager, EXEC_PROFILE_DEFAULT, - ExecutionProfile) + ExecutionProfile, + ControlConnectionQueryFallback) from cassandra.pool import Host from cassandra.connection import (ConnectionException, EndPoint, DefaultEndPoint, DefaultEndPointFactory, UnixSocketEndPoint) -from cassandra.policies import (SimpleConvictionPolicy, RoundRobinPolicy, - ConstantReconnectionPolicy, IdentityTranslator) +from cassandra.policies import (HostDistance, SimpleConvictionPolicy, + RoundRobinPolicy, ConstantReconnectionPolicy, + IdentityTranslator) PEER_IP = "foobar" @@ -230,6 +232,21 @@ def _refresh_control_connection_over_network(self): self.connection.original_endpoint = self.connection.endpoint self.control_connection.refresh_node_list_and_token_map() + def _use_cluster_down_handling( + self, sessions=(), + fallback=ControlConnectionQueryFallback.Disabled): + self.cluster.sessions = list(sessions) + self.cluster._discount_down_events = True + self.cluster.allow_control_connection_query_fallback = fallback + self.cluster.profile_manager = Mock() + self.cluster.profile_manager.distance.return_value = \ + HostDistance.LOCAL + # The real method reports whether the executor accepted the work. + self.cluster.on_down_potentially_blocking = Mock(return_value=True) + self.cluster.on_down = Cluster.on_down.__get__(self.cluster) + self.cluster.signal_connection_failure = \ + Cluster.signal_connection_failure.__get__(self.cluster) + def test_wait_for_schema_agreement(self): """ Basic test with all schema versions agreeing @@ -469,6 +486,161 @@ def test_schema_query_uses_shard_aware_connection_original_endpoint(self): assert query == self.control_connection._SELECT_SCHEMA_PEERS_TEMPLATE \ .format(nt_col_name='rpc_address') + def test_defunct_tcp_control_reconnects_when_open_pool_discounts_down(self): + host = self.cluster.metadata.get_host_by_host_id('uuid1') + host.set_up() + session = Mock() + session.get_pool_state.return_value = { + host: {'open_count': 1}} + self._use_cluster_down_handling([session]) + self.connection.is_defunct = True + self.connection.last_error = ConnectionException( + 'control connection failed') + self.cluster.executor.reset_mock() + + self.control_connection._signal_error() + + assert host.is_up is True + self.cluster.on_down_potentially_blocking.assert_not_called() + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_defunct_control_reconnects_when_conviction_is_rejected(self): + host = self.cluster.metadata.get_host_by_host_id('uuid1') + host.set_up() + self._use_cluster_down_handling() + self.connection.is_defunct = True + self.connection.last_error = OperationTimedOut() + self.cluster.executor.reset_mock() + + self.control_connection._signal_error() + + assert host.is_up is True + self.cluster.on_down_potentially_blocking.assert_not_called() + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_defunct_control_reconnects_when_host_is_already_down(self): + host = self.cluster.metadata.get_host_by_host_id('uuid1') + host.set_down() + self._use_cluster_down_handling() + self.connection.is_defunct = True + self.connection.last_error = ConnectionException( + 'control connection failed') + self.cluster.executor.reset_mock() + + self.control_connection._signal_error() + + self.cluster.on_down_potentially_blocking.assert_not_called() + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_defunct_control_reconnects_when_host_reconnector_is_active(self): + host = self.cluster.metadata.get_host_by_host_id('uuid1') + host.set_up() + host.get_and_set_reconnection_handler(Mock()) + self._use_cluster_down_handling() + self.connection.is_defunct = True + self.connection.last_error = ConnectionException( + 'control connection failed') + self.cluster.executor.reset_mock() + + self.control_connection._signal_error() + + assert host.is_up is False + self.cluster.on_down_potentially_blocking.assert_not_called() + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_defunct_control_reconnects_when_pool_creation_is_disabled(self): + host = self.cluster.metadata.get_host_by_host_id('uuid1') + host.set_up() + self._use_cluster_down_handling( + fallback=ControlConnectionQueryFallback.SkipPoolCreation) + self.connection.is_defunct = True + self.connection.last_error = ConnectionException( + 'control connection failed') + self.cluster.executor.reset_mock() + + self.control_connection._signal_error() + + assert host.is_up is True + self.cluster.on_down_potentially_blocking.assert_not_called() + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_defunct_control_waits_for_dispatched_down_callback(self): + host = self.cluster.metadata.get_host_by_host_id('uuid1') + host.set_up() + self._use_cluster_down_handling() + self.connection.is_defunct = True + self.connection.last_error = ConnectionException( + 'control connection failed') + self.cluster.executor.reset_mock() + + self.control_connection._signal_error() + + assert host.is_up is False + self.cluster.on_down_potentially_blocking.assert_called_once_with( + host, False) + self.cluster.executor.submit.assert_not_called() + + self.control_connection.on_down(host) + + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_signal_error_reconnects_non_defunct_connection(self): + self.connection.is_defunct = False + self.cluster.executor.reset_mock() + + self.control_connection._signal_error() + + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_signal_error_reconnects_when_host_is_unresolved(self): + self._forget_local_host() + self.connection.is_defunct = True + self.connection.last_error = ConnectionException( + 'control connection failed') + self.cluster.executor.reset_mock() + + self.control_connection._signal_error() + + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_signal_error_does_nothing_after_control_connection_shutdown(self): + self.control_connection._is_shutdown = True + self.connection.is_defunct = True + self.cluster.executor.reset_mock() + + self.control_connection._signal_error() + + self.cluster.executor.submit.assert_not_called() + + def test_defunct_control_reconnects_when_down_dispatch_is_dropped(self): + # on_down() marks the host down but the executor refuses the DOWN + # callback, so nothing else will reconnect the control connection. + host = self.cluster.metadata.get_host_by_host_id('uuid1') + host.set_up() + self._use_cluster_down_handling() + self.cluster.on_down_potentially_blocking = \ + Cluster.on_down_potentially_blocking.__get__(self.cluster) + self.connection.is_defunct = True + self.connection.last_error = ConnectionException( + 'control connection failed') + self.cluster.executor.reset_mock() + self.cluster.executor.submit.side_effect = [ + RuntimeError('cannot schedule new futures'), Mock()] + + self.control_connection._signal_error() + + assert host.is_up is False + assert self.cluster.executor.submit.call_args_list[-1] == call( + self.control_connection._reconnect) + def test_refresh_network_local_preserves_known_unix_endpoint(self): maintenance_endpoint, local_host = \ self._discover_local_host_over_unix() @@ -486,7 +658,7 @@ def test_refresh_network_local_preserves_known_unix_endpoint(self): self.connection.last_error = connection_error # Model a conviction whose DOWN transition is discounted because a # usable session pool remains: no control on_down callback is queued. - self.cluster.signal_connection_failure = Mock(return_value=True) + self.cluster.signal_connection_failure = Mock(return_value=False) self.cluster.executor.reset_mock() self.control_connection._signal_error() @@ -501,7 +673,7 @@ def test_unix_signal_error_reconnects_if_down_notification_suppressed(self): connection_error = ConnectionException('control connection failed') self.connection.is_defunct = True self.connection.last_error = connection_error - self.cluster.signal_connection_failure = Mock(return_value=True) + self.cluster.signal_connection_failure = Mock(return_value=False) self.cluster.executor.reset_mock() self.control_connection._signal_error() @@ -519,7 +691,7 @@ def test_tcp_route_mismatch_reconnects_if_down_notification_suppressed(self): connection_error = ConnectionException('control connection failed') self.connection.is_defunct = True self.connection.last_error = connection_error - self.cluster.signal_connection_failure = Mock(return_value=True) + self.cluster.signal_connection_failure = Mock(return_value=False) self.cluster.executor.reset_mock() self.control_connection._signal_error() @@ -564,7 +736,7 @@ def test_route_mismatch_signal_error_reconnects_if_host_already_down(self): connection_error = ConnectionException('control connection failed') self.connection.is_defunct = True self.connection.last_error = connection_error - self.cluster.signal_connection_failure = Mock(return_value=True) + self.cluster.signal_connection_failure = Mock(return_value=False) self.cluster.executor.reset_mock() self.control_connection._signal_error() @@ -582,12 +754,7 @@ def test_route_mismatch_signal_error_reconnects_if_host_reconnecting(self): self.connection.is_defunct = True self.connection.last_error = connection_error - def transition_without_notification(host, *_args, **_kwargs): - host.set_down() - return True - - self.cluster.signal_connection_failure = Mock( - side_effect=transition_without_notification) + self.cluster.signal_connection_failure = Mock(return_value=False) self.cluster.executor.reset_mock() self.control_connection._signal_error() From 8b5a73262d91b00729db644f4faf1a4aa05989f8 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Mon, 21 Sep 2026 21:50:37 -0400 Subject: [PATCH 2/4] control-connection: preserve reconnect ownership The direct recovery path can overlap queued attempts, heartbeat returns, topology-driven replacements, and finite retry schedules. Without a single owner for pending and delayed work, later triggers can reset backoff, stale handlers can replace healthy connections, and exhausted handlers can suppress an independent failure. Serialize reconnect submissions, retain triggers until an active attempt or handler owns them, and release handler slots with identity checks. Preserve failures that arrive during proactive retries, honor empty and exhausted schedules, and keep a successfully restored control connection open while still closing host-probe connections. Make connection installation and shutdown races explicit so an overlapping failed attempt cannot park retries over a newer connection and an orphaned connection is closed. --- cassandra/cluster.py | 404 +++++++++++++-- cassandra/pool.py | 14 +- tests/unit/test_cluster.py | 37 ++ tests/unit/test_control_connection.py | 676 ++++++++++++++++++++++++-- 4 files changed, 1056 insertions(+), 75 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 3b2bec1ce4..0df2f1621f 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -2018,8 +2018,25 @@ def _start_reconnector(self, host, is_host_addition): @run_in_executor def on_down_potentially_blocking(self, host, is_host_addition): - self.profile_manager.on_down(host) - self.control_connection.on_down(host) + # The load balancing policies go first, so that the query plan the + # control connection's reconnect walks no longer offers the host that + # just went down; otherwise the reconnect submitted below can start + # before the policies have dropped it and burn a connect_timeout on a + # dead node. Both calls are guarded: on_down() reports back that DOWN + # handling was dispatched, and callers rely on that to mean a + # reconnection will be started, so a user policy or listener raising + # must not be what stops it from happening. By the same token, a + # failure in the control connection -- the executor rejecting the + # reconnect submission, say -- must not skip the pool, listener and + # reconnector work below. + try: + self.profile_manager.on_down(host) + except Exception: + log.exception("Error in load balancing policy down handler for host %s", host) + try: + self.control_connection.on_down(host) + except Exception: + log.exception("Error in control connection down handler for host %s", host) for session in tuple(self.sessions): session.on_down(host) @@ -3731,23 +3748,125 @@ class _ControlReconnectionHandler(_ReconnectionHandler): Internal """ - def __init__(self, control_connection, *args, **kwargs): - _ReconnectionHandler.__init__(self, *args, **kwargs) + # on_reconnection() installs the connection as the control connection, + # so it must outlive the handler run that opened it. + _keeps_connection = True + + def __init__(self, control_connection, scheduler, schedule): + # The run-completed callback releases this handler's own slot. It is + # identity-checked, so a run that finishes after another thread has + # installed a replacement leaves that replacement alone. Anything that + # evicts a handler must also cancel it, or the evicted one keeps + # retrying where nothing can find it. + _ReconnectionHandler.__init__(self, scheduler, schedule, self._release) self.control_connection = weakref.proxy(control_connection) + # Remember a failed connection that caused this retry run. If the + # finite schedule is exhausted, the heartbeat will keep returning this + # same object and must not start the schedule over from the beginning. + # A healthy connection may also be replaced after a topology event; + # exhausting that attempt must not suppress recovery if the connection + # fails later for an independent reason. + connection = control_connection._connection + self._failed_connection = connection \ + if connection is not None and ( + connection.is_defunct or connection.is_closed) else None + # A reconnect trigger that arrives while an attempt is actually + # running is normally covered by that attempt. If this turns out to be + # the final failed attempt, though, there is no later retry to own the + # trigger, so _release() must hand it back to reconnect(). + self._is_running = False + self._reconnect_requested = False + # A proactive handler may be waiting while its still-healthy control + # connection fails. Its remaining retries own the immediate recovery, + # but the failure must receive a fresh schedule if those retries are + # exhausted. Unlike a trigger received during a running attempt, this + # state therefore survives every non-final handler run. + self._new_failure_detected = False + + def run(self): + try: + control_connection = self.control_connection + with control_connection._reconnection_lock: + self._is_running = True + except ReferenceError: + return _ReconnectionHandler.run(self) + + try: + _ReconnectionHandler.run(self) + finally: + try: + with control_connection._reconnection_lock: + self._is_running = False + except ReferenceError: + pass def try_reconnect(self): return self.control_connection._reconnect_internal() def on_reconnection(self, connection): - self.control_connection._set_new_connection(connection) + try: + # Resolving the attribute is what dereferences the weak proxy, so + # it is bound here rather than called directly: a ReferenceError + # raised from inside _set_new_connection() would mean the + # connection was already adopted, and must not be closed. + set_new_connection = self.control_connection._set_new_connection + except ReferenceError: + # The ControlConnection was collected while we were retrying. The + # handler has already marked the connection as handed off, so + # nothing else would ever close it. + connection.close() + return + set_new_connection(connection) def on_exception(self, exc, next_delay): - # TODO only overridden to add logging, so add logging - if isinstance(exc, AuthenticationFailed): - return False + # Every reason to stop retrying is worth retrying here: an attempt + # covers the whole query plan, so an authentication failure is a + # failure of the host it happened to reach, not of the cluster. + log.debug("Error trying to reconnect control connection: %r", exc) + + if next_delay is None: + # The schedule is exhausted, so this handler will never run again. + # Release the slot it occupies while remembering the connection + # whose retries were exhausted. A later explicit error may start a + # fresh run, but heartbeats for this same defunct connection must + # not reset the finite schedule on every interval. + self._release(exhausted=True) else: - log.debug("Error trying to reconnect control connection: %r", exc) - return True + # This failure already has another scheduled attempt to own any + # trigger that arrived while it was running. + try: + control_connection = self.control_connection + with control_connection._reconnection_lock: + self._is_running = False + self._reconnect_requested = False + except ReferenceError: + pass + + return True + + def _release(self, exhausted=False): + reconnect_requested = False + try: + control_connection = self.control_connection + with control_connection._reconnection_lock: + if control_connection._reconnection_handler is self: + control_connection._reconnection_handler = None + reconnect_requested = exhausted and ( + self._reconnect_requested or + self._new_failure_detected) + self._reconnect_requested = False + self._new_failure_detected = False + if (exhausted and not reconnect_requested and + self._failed_connection is not None and + control_connection._connection is + self._failed_connection): + control_connection._reconnection_exhausted_connection = \ + self._failed_connection + except ReferenceError: + pass # our weak reference to the ControlConnection is no good + + if reconnect_requested: + control_connection.reconnect() def _watch_callback(obj_weakref, method_name, *args, **kwargs): @@ -3844,6 +3963,24 @@ def __init__(self, cluster, timeout, self._reconnection_handler = None self._reconnection_lock = RLock() + # A finite retry schedule may be exhausted while the heartbeat still + # owns and returns the same defunct connection. Keep that identity so + # heartbeat passes do not create a fresh schedule indefinitely. + self._reconnection_exhausted_connection = None + self._reconnect_pending = False + # Set when reconnect() is called while an attempt is already pending. + # If that attempt fails without installing a retry handler, its final + # action consumes this flag by queuing one follow-up attempt. + self._reconnect_requested = False + # Bumped every time _reconnect_pending is raised. An attempt clears the + # flag only while it still owns it: _set_new_connection() drops the flag + # as the connection goes in, so a newer reconnect() can claim it before + # the attempt that installed the connection reaches its finally clause, + # and clearing it there would strand the attempt that newer flag owns. + self._reconnect_pending_seq = 0 + # Bumped every time a connection is installed. _reconnect() compares it + # across its attempt to tell whether another attempt got there first. + self._connection_generation = 0 self._event_schedule_times = {} @@ -3854,20 +3991,61 @@ def connect(self): self._protocol_version = self._cluster.protocol_version self._set_new_connection(self._reconnect_internal()) - self._cluster.metadata.dbaas = self._connection._product_type == dscloud.DATASTAX_CLOUD_PRODUCT_TYPE + # _set_new_connection declines to install anything once shutdown() has + # run, so there may be no connection to ask. + if self._connection: + self._cluster.metadata.dbaas = self._connection._product_type == dscloud.DATASTAX_CLOUD_PRODUCT_TYPE def _set_new_connection(self, conn): """ - Replace existing connection (if there is one) and close it. + Adopt `conn` as the control connection, closing the one it replaces. + + Also ends any reconnection in progress: a handler parked in the slot is + cancelled and cleared, because it would otherwise be mistaken for one + still retrying and its next attempt would replace this connection. + + If the ControlConnection has already been shut down, `conn` is not + installed at all -- nothing would ever use it or close it on our + behalf, so it is closed here instead. """ - with self._lock: - old = self._connection - self._connection = conn + # Whatever put this connection in place, the reconnection is over. A + # handler left parked in the slot would be mistaken for one still + # retrying, and its next attempt would replace this connection. + # + # _reconnection_lock is held across the install so that the generation + # bump is visible to any _reconnect() that is about to decide whether + # another attempt beat it; releasing it first leaves a window where a + # failing attempt reads the old generation and parks a handler over + # this healthy connection. + with self._reconnection_lock: + if self._reconnection_handler: + self._reconnection_handler.cancel() + self._reconnection_handler = None + + # A connection is in place, so an error on it must be able to queue + # a fresh attempt rather than be collapsed into the one ending here. + self._reconnect_pending = False + self._reconnect_requested = False + + with self._lock: + if self._is_shutdown: + # shutdown() won the race, so nothing would ever use this + # connection or close it on our behalf. + old, orphan = None, conn + else: + old, orphan = self._connection, None + self._connection = conn + self._reconnection_exhausted_connection = None + self._connection_generation += 1 if old: log.debug("[control connection] Closing old connection %r, replacing with %r", old, conn) old.close() + if orphan: + log.debug("[control connection] Control connection is shut down, closing new connection %r", orphan) + orphan.close() + def _try_connect_to_hosts(self): errors = {} @@ -4004,44 +4182,177 @@ def reconnect(self): if self._is_shutdown: return - self._submit(self._reconnect) + # Collapse attempts that are queued but have not started yet. Without + # this, a burst of errors queues one _reconnect() each, and every one + # of them past the first cancels the reconnection handler the previous + # one installed and restarts its backoff schedule. + # + # An in-flight reconnection handler is already retrying on its own + # schedule, and _reconnect() would cancel it and restart that schedule + # from its initial delay. The check lives here rather than in the + # callers so that it covers every entry point: return_connection() is + # driven by the heartbeat and fires once per idle_heartbeat_interval + # for as long as the control connection stays defunct, which would + # otherwise reset the backoff on every pass and stop it ever growing. + # The lock is an RLock, so callers already holding it re-enter safely. + with self._reconnection_lock: + if self._reconnection_handler is not None: + handler = self._reconnection_handler + if getattr(handler, '_is_running', False) is True: + handler._reconnect_requested = True + + # A handler can have been started proactively while the + # current connection was healthy. If that connection fails + # during the handler's backoff, the remaining proactive retry + # still goes first, but exhaustion must hand this independent + # failure a fresh schedule. Record it once so recurring + # heartbeats for an already-failed connection do not extend a + # finite schedule indefinitely. + connection = self._connection + if (getattr(handler, '_failed_connection', None) is None and + connection is not None and + (connection.is_defunct or connection.is_closed)): + handler._failed_connection = connection + handler._new_failure_detected = True + log.debug("[control connection] Reconnection already in progress, " + "not starting another one") + return + if self._reconnect_pending: + log.debug("[control connection] A reconnection attempt is " + "already queued") + # Do not discard this trigger. The pending attempt may fail + # before it installs a handler that owns future retries. + self._reconnect_requested = True + return + pending_seq = self._raise_reconnect_pending() + + submitted = None + try: + submitted = self._submit(self._reconnect) + finally: + if submitted is None: + # Nothing was queued, so nothing will clear the flag. This has + # to hold even when the submission raised, or no further + # reconnection would ever be attempted. + self._clear_reconnect_pending(pending_seq) + + def _raise_reconnect_pending(self): + """ + Mark a reconnection attempt as pending and return a token identifying + it. Only the holder of the newest token may clear the flag again. + """ + with self._reconnection_lock: + self._reconnect_pending = True + self._reconnect_pending_seq += 1 + return self._reconnect_pending_seq + + def _clear_reconnect_pending(self, pending_seq): + """ + Clear the pending flag, unless a newer attempt has claimed it since + `pending_seq` was handed out -- that attempt is the one the flag now + stands for, and dropping it would let a burst of errors queue several + concurrent reconnects. + """ + with self._reconnection_lock: + if self._reconnect_pending_seq == pending_seq: + self._reconnect_pending = False + self._reconnect_requested = False + + def _finish_reconnect(self, pending_seq): + """ + Finish an active attempt, preserving any trigger it collapsed unless + a connection or reconnection handler now owns future work. + """ + follow_up_seq = None + with self._reconnection_lock: + if self._reconnect_pending_seq != pending_seq: + return + + if (self._reconnection_handler is not None or + not self._reconnect_requested): + self._reconnect_pending = False + self._reconnect_requested = False + return + + # Keep the pending flag raised while handing the retained trigger + # to the executor, so another burst still collapses into this one. + self._reconnect_requested = False + self._reconnect_pending_seq += 1 + follow_up_seq = self._reconnect_pending_seq + + submitted = None + try: + submitted = self._submit(self._reconnect) + finally: + if submitted is None: + self._clear_reconnect_pending(follow_up_seq) def _reconnect(self): + # _reconnect_pending stays set for as long as this attempt runs. + # _reconnect_internal() walks the whole query plan twice with a DNS + # re-resolution in between, which routinely outlasts + # idle_heartbeat_interval when the cluster is unreachable; clearing the + # flag here would let every heartbeat in that window queue another + # attempt, each one cancelling the handler the previous one installed + # and restarting its backoff from the initial delay. + pending_seq = self._raise_reconnect_pending() + + with self._lock: + generation = self._connection_generation + log.debug("[control connection] Attempting to reconnect") try: self._set_new_connection(self._reconnect_internal()) - except NoHostAvailable: + except (NoHostAvailable, UnresolvableContactPoints): # make a retry schedule (which includes backoff) schedule = self._cluster.reconnection_policy.new_schedule() with self._reconnection_lock: + with self._lock: + if self._connection_generation != generation: + # An attempt that overlapped this one installed a + # connection while we were failing. Parking a handler + # now would block every later reconnect() for the whole + # backoff and then replace a healthy connection. + log.debug("[control connection] Reconnect failed but " + "another attempt succeeded, not scheduling " + "retries") + return # cancel existing reconnection attempts if self._reconnection_handler: self._reconnection_handler.cancel() # when a connection is successfully made, _set_new_connection - # will be called with the new connection and then our - # _reconnection_handler will be cleared out - self._reconnection_handler = _ControlReconnectionHandler( - self, self._cluster.scheduler, schedule, - self._get_and_set_reconnection_handler, - new_handler=None) - self._reconnection_handler.start() + # will be called with the new connection and will clear out + # our _reconnection_handler + handler = _ControlReconnectionHandler( + self, self._cluster.scheduler, schedule) + self._reconnection_handler = handler + try: + handler.start() + except StopIteration: + # An empty schedule means that the policy permits no + # retries. Treat it like a finite schedule exhausted by a + # failed handler run so heartbeats for the same defunct + # connection do not start it over indefinitely. + handler._release(exhausted=True) + except Exception: + # A handler that never started would sit in the slot + # forever, and reconnect() would take it for one still + # retrying. + if self._reconnection_handler is handler: + self._reconnection_handler = None + raise except Exception: log.debug("[control connection] error reconnecting", exc_info=True) raise - - def _get_and_set_reconnection_handler(self, new_handler): - """ - Called by the _ControlReconnectionHandler when a new connection - is successfully created. Clears out the _reconnection_handler on - this ControlConnection. - """ - with self._reconnection_lock: - old = self._reconnection_handler - self._reconnection_handler = new_handler - return old + finally: + # The attempt is over either way: a connection is installed (the + # flag was already cleared as it went in), a handler is parked and + # collapses later attempts on its own, or a retained trigger queues + # one follow-up after a failure that left no durable retry work. + self._finish_reconnect(pending_seq) def _submit(self, *args, **kwargs): try: @@ -4608,7 +4919,8 @@ def _signal_error(self): return # If the connection is not defunct, the host is unresolved, or DOWN - # handling was suppressed, reconnect manually. + # handling was suppressed, reconnect manually. reconnect() leaves an + # in-flight reconnection handler alone on its own schedule. self.reconnect() def on_up(self, host): @@ -4617,12 +4929,13 @@ def on_up(self, host): def on_down(self, host): conn = self._connection - if self._connection_matches_host(conn, host) and \ - self._reconnection_handler is None: - log.debug("[control connection] Control connection host (%s) is " - "considered down, starting reconnection", host) - # this will result in a task being submitted to the executor to reconnect - self.reconnect() + if not self._connection_matches_host(conn, host): + return + + log.debug("[control connection] Control connection host (%s) is " + "considered down, starting reconnection", host) + # this will result in a task being submitted to the executor to reconnect + self.reconnect() def on_add(self, host, refresh_nodes=True): if refresh_nodes: @@ -4643,7 +4956,12 @@ def get_connections(self): def return_connection(self, connection): if connection is self._connection and (connection.is_defunct or connection.is_closed): - self.reconnect() + with self._reconnection_lock: + if connection is self._reconnection_exhausted_connection: + log.debug("[control connection] Reconnection schedule is " + "exhausted for the defunct connection") + return + self.reconnect() def _stop_scheduler(scheduler, thread): diff --git a/cassandra/pool.py b/cassandra/pool.py index 14829ffa26..6813ca0746 100644 --- a/cassandra/pool.py +++ b/cassandra/pool.py @@ -264,6 +264,11 @@ class _ReconnectionHandler(object): _cancelled = False + # Whether on_reconnection() keeps the connection it is handed. A handler + # that only uses it to probe the host leaves this False and run() closes + # the connection for it. + _keeps_connection = False + def __init__(self, scheduler, schedule, callback, *callback_args, **callback_kwargs): self.scheduler = scheduler self.schedule = schedule @@ -284,6 +289,7 @@ def run(self): return conn = None + handed_off = False try: conn = self.try_reconnect() except Exception as exc: @@ -303,10 +309,16 @@ def run(self): self.scheduler.schedule(next_delay, self.run) else: if not self._cancelled: + # Mark the handoff before it happens: on_reconnection() adopts + # the connection and may then raise (installing the new control + # connection closes the old one, which runs user callbacks). If + # the flag were set afterwards, that raise would leave us + # closing a connection the subclass is already using. + handed_off = self._keeps_connection self.on_reconnection(conn) self.callback(*(self.callback_args), **(self.callback_kwargs)) finally: - if conn: + if conn and not handed_off: conn.close() def cancel(self): diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 4fd6f30437..36302ba030 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -785,6 +785,43 @@ def setUp(self): "127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4()) self.host.set_up() + def test_down_handling_reaches_control_connection_despite_a_raising_policy(self): + # signal_connection_failure() reports DOWN handling as dispatched, and + # the control connection relies on that to mean its own reconnection + # will be started. A user policy raising must not be what prevents it. + del self.cluster.on_down_potentially_blocking # use the real method + self.cluster.profile_manager.on_down = Mock( + side_effect=RuntimeError('load balancing policy failed')) + self.cluster.control_connection = Mock() + self.cluster._start_reconnector = Mock() + # The method runs in the executor; call the body directly so nothing + # is swallowed by the future. + body = Cluster.on_down_potentially_blocking.__wrapped__ + + body(self.cluster, self.host, is_host_addition=False) + + self.cluster.control_connection.on_down.assert_called_once_with( + self.host) + self.cluster._start_reconnector.assert_called_once() + + def test_down_handling_continues_when_the_control_connection_raises(self): + # The control connection goes first, so a failure there -- the executor + # rejecting its reconnect submission, say -- must not skip the pool and + # listener work or the host reconnector that follows it. + del self.cluster.on_down_potentially_blocking # use the real method + self.cluster.control_connection = Mock() + self.cluster.control_connection.on_down.side_effect = RuntimeError( + 'cannot schedule new futures after shutdown') + self.cluster._start_reconnector = Mock() + self.cluster.profile_manager.on_down = Mock() + body = Cluster.on_down_potentially_blocking.__wrapped__ + + body(self.cluster, self.host, is_host_addition=False) + + self.cluster.profile_manager.on_down.assert_called_once_with(self.host) + self.cluster._start_reconnector.assert_called_once_with( + self.host, False) + def test_signal_connection_failure_rejected_by_conviction_policy(self): error = ConnectionException("connection failed") self.host.signal_connection_failure = Mock(return_value=False) diff --git a/tests/unit/test_control_connection.py b/tests/unit/test_control_connection.py index b4a9224360..101c6f71ac 100644 --- a/tests/unit/test_control_connection.py +++ b/tests/unit/test_control_connection.py @@ -13,21 +13,26 @@ # limitations under the License. import unittest +import weakref from concurrent.futures import ThreadPoolExecutor from unittest.mock import Mock, ANY, call, patch -from cassandra import OperationTimedOut, SchemaTargetType, SchemaChangeType +from cassandra import (AuthenticationFailed, OperationTimedOut, + SchemaTargetType, SchemaChangeType, + UnresolvableContactPoints) from cassandra.protocol import ResultMessage, RESULT_KIND_ROWS from cassandra.cluster import (Cluster, ControlConnection, _Scheduler, ProfileManager, EXEC_PROFILE_DEFAULT, ExecutionProfile, - ControlConnectionQueryFallback) -from cassandra.pool import Host + ControlConnectionQueryFallback, + NoHostAvailable, _ControlReconnectionHandler) +from cassandra.pool import Host, _ReconnectionHandler from cassandra.connection import (ConnectionException, EndPoint, DefaultEndPoint, DefaultEndPointFactory, UnixSocketEndPoint) from cassandra.policies import (HostDistance, SimpleConvictionPolicy, RoundRobinPolicy, ConstantReconnectionPolicy, + ExponentialReconnectionPolicy, IdentityTranslator) PEER_IP = "foobar" @@ -151,6 +156,7 @@ def _node_meta_results(local_results, peer_results): class MockConnection(object): is_defunct = False + is_closed = False def __init__(self): self.endpoint = DefaultEndPoint("192.168.1.0") @@ -247,6 +253,13 @@ def _use_cluster_down_handling( self.cluster.signal_connection_failure = \ Cluster.signal_connection_failure.__get__(self.cluster) + def _discount_down_for(self, host): + """Model a session pool that keeps ``host`` up despite a conviction.""" + session = Mock() + session.get_pool_state.return_value = {host: {'open_count': 1}} + self._use_cluster_down_handling([session]) + return session + def test_wait_for_schema_agreement(self): """ Basic test with all schema versions agreeing @@ -620,6 +633,601 @@ def test_signal_error_does_nothing_after_control_connection_shutdown(self): self.cluster.executor.submit.assert_not_called() + def test_signal_error_leaves_an_in_flight_reconnection_alone(self): + # _reconnect() cancels the handler and restarts its schedule from the + # initial delay, so repeated errors must not keep resetting the backoff. + host = self.cluster.metadata.get_host_by_host_id('uuid1') + host.set_down() + self._use_cluster_down_handling() + self.control_connection._reconnection_handler = Mock() + self.connection.is_defunct = True + self.connection.last_error = ConnectionException( + 'control connection failed') + self.cluster.executor.reset_mock() + + self.control_connection._signal_error() + + self.cluster.executor.submit.assert_not_called() + + def _make_reconnection_handler(self): + handler = _ControlReconnectionHandler( + self.control_connection, self.cluster.scheduler, iter([1.0])) + self.control_connection._reconnection_handler = handler + return handler + + def test_reconnection_handler_releases_its_slot_when_it_gives_up(self): + handler = self._make_reconnection_handler() + + handler.on_exception(ConnectionException('refused'), None) + + assert self.control_connection._reconnection_handler is None + + def test_reconnection_handler_keeps_its_slot_while_it_retries(self): + for exc in (ConnectionException('refused'), + AuthenticationFailed('bad password')): + with self.subTest(exc=exc): + handler = self._make_reconnection_handler() + + assert handler.on_exception(exc, 1.0) + + assert self.control_connection._reconnection_handler is handler + + def test_reconnection_handler_never_releases_a_replacement(self): + handler = self._make_reconnection_handler() + replacement = self._make_reconnection_handler() + + handler.on_exception(ConnectionException('refused'), None) + + assert self.control_connection._reconnection_handler is replacement + + def test_reconnection_handler_run_leaves_a_replacement_in_the_slot(self): + # A finishing handler must not clear a replacement that another thread + # parked in the slot while it was handing its connection over. + # _set_new_connection() has already released this handler by then, so + # clearing again can only evict someone else -- and it would do so + # without cancelling them, leaving a handler that keeps retrying where + # reconnect() can no longer see it. + handler = self._make_reconnection_handler() + self.control_connection._connection = None + conn = Mock() + replacement = [] + + def install_and_lose_the_race(new_conn): + # What _set_new_connection() really does -- release the slot and + # adopt the connection -- plus another thread winning the race to + # park a fresh handler in the slot it just emptied. + self.control_connection._reconnection_handler = None + self.control_connection._connection = new_conn + replacement.append(self._make_reconnection_handler()) + + with patch.object(handler, 'try_reconnect', return_value=conn), \ + patch.object(self.control_connection, '_set_new_connection', + side_effect=install_and_lose_the_race): + handler.run() + + assert self.control_connection._reconnection_handler is replacement[0] + + def test_reconnection_handler_keeps_the_connection_it_installs(self): + # run() closes the connection it opened, which is right for the host + # handler that only probes with it, but this one hands it to the + # control connection. + handler = self._make_reconnection_handler() + self.control_connection._connection = None + conn = Mock() + + with patch.object(handler, 'try_reconnect', return_value=conn): + handler.run() + + assert self.control_connection._connection is conn + conn.close.assert_not_called() + + def test_reconnection_handler_keeps_the_connection_a_failed_install_took(self): + # Installing the new control connection closes the old one, which runs + # user callbacks, and one of those may raise. The connection is already + # adopted by then, so it must not be closed on the way out. + handler = self._make_reconnection_handler() + self.control_connection._connection = None + conn = Mock() + + with patch.object(handler, 'try_reconnect', return_value=conn), \ + patch.object(self.control_connection, '_set_new_connection', + side_effect=RuntimeError('listener failed')): + with self.assertRaises(RuntimeError): + handler.run() + + conn.close.assert_not_called() + + def test_reconnection_handler_closes_a_connection_it_only_probes_with(self): + # The plain handler, like the host one, only uses the connection to + # prove the host is reachable, so run() still closes it for it. + handler = _ReconnectionHandler(self.cluster.scheduler, iter([1.0]), + Mock()) + conn = Mock() + + with patch.object(handler, 'try_reconnect', return_value=conn): + handler.run() + + conn.close.assert_called_once_with() + + def test_reconnecting_successfully_releases_a_parked_handler(self): + # A handler installed by an earlier failure is still backing off when + # an unrelated reconnect succeeds. Left in the slot it would look like + # a reconnection in progress to every later error. + handler = self._make_reconnection_handler() + self.control_connection._connection = None + + with patch.object(self.control_connection, '_reconnect_internal', + return_value=Mock()): + self.control_connection._reconnect() + + assert self.control_connection._reconnection_handler is None + assert handler._cancelled + + def test_set_new_connection_closes_a_connection_shutdown_beat(self): + # shutdown() already closed the control connection, so nothing would + # ever use this one or close it on our behalf. + self.control_connection._is_shutdown = True + conn = Mock() + + self.control_connection._set_new_connection(conn) + + conn.close.assert_called_once_with() + assert self.control_connection._connection is self.connection + + def test_signal_error_reconnects_once_a_reconnection_has_given_up(self): + host = self.cluster.metadata.get_host_by_host_id('uuid1') + host.set_down() + self._use_cluster_down_handling() + handler = self._make_reconnection_handler() + handler.on_exception(ConnectionException('refused'), None) + self.connection.is_defunct = True + self.connection.last_error = ConnectionException( + 'control connection failed') + self.cluster.executor.reset_mock() + + self.control_connection._signal_error() + + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_reconnect_collapses_an_attempt_that_has_not_started(self): + self.cluster.executor.reset_mock() + + self.control_connection.reconnect() + self.control_connection.reconnect() + + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_reconnect_is_queued_again_once_the_attempt_starts(self): + self.cluster.executor.reset_mock() + self.control_connection.reconnect() + self.control_connection._connection = None + + with patch.object(self.control_connection, '_reconnect_internal', + return_value=Mock()): + self.control_connection._reconnect() + + self.control_connection.reconnect() + + assert self.cluster.executor.submit.call_args_list == [ + call(self.control_connection._reconnect), + call(self.control_connection._reconnect)] + + def test_reconnect_does_not_clear_a_flag_a_newer_attempt_claimed(self): + # _set_new_connection() drops the pending flag as the connection goes + # in. If that connection errors before the installing attempt returns, + # the reconnect() it triggers claims the flag, and the finishing + # attempt must not clear it -- a further trigger would then queue a + # second attempt that runs alongside the first. + self.cluster.executor.reset_mock() + self.control_connection._connection = None + install = self.control_connection._set_new_connection + + def install_then_lose_the_connection(conn): + install(conn) + self.control_connection.reconnect() + + with patch.object(self.control_connection, '_reconnect_internal', + return_value=Mock()), \ + patch.object(self.control_connection, '_set_new_connection', + side_effect=install_then_lose_the_connection): + self.control_connection._reconnect() + + assert self.control_connection._reconnect_pending + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + self.control_connection.reconnect() + + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_reconnect_collapses_an_attempt_while_one_is_running(self): + # _reconnect_internal() walks the whole query plan and can outlast the + # idle heartbeat, which calls reconnect() through return_connection(). + # Those calls must not queue a second attempt that would cancel the + # first one's handler and restart its backoff. + def reconnect_while_running(): + self.cluster.executor.reset_mock() + self.control_connection.reconnect() + raise NoHostAvailable('no host', {}) + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=reconnect_while_running): + self.control_connection._reconnect() + + self.cluster.executor.submit.assert_not_called() + + def test_reconnect_handler_owns_trigger_after_dns_failure(self): + # A trigger arriving during an attempt must not queue duplicate work + # when that attempt leaves a backoff handler owning future retries. + def reconnect_while_running(): + self.cluster.executor.reset_mock() + self.control_connection.reconnect() + raise UnresolvableContactPoints({}) + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=reconnect_while_running): + self.control_connection._reconnect() + + self.cluster.executor.submit.assert_not_called() + assert self.control_connection._reconnection_handler is not None + assert not self.control_connection._reconnect_pending + + def test_reconnect_preserves_trigger_if_new_handler_already_exhausted(self): + # A zero-delay finite handler can exhaust on another worker before the + # attempt that started it reaches _finish_reconnect(). It no longer + # owns retries at that point, so a trigger collapsed in between must + # be submitted as follow-up work. + self.cluster.reconnection_policy = ConstantReconnectionPolicy( + 0, max_attempts=1) + self.connection.is_defunct = True + + def exhaust_then_trigger(_delay, run): + run() + self.control_connection.reconnect() + + self.cluster.scheduler.schedule.side_effect = exhaust_then_trigger + self.cluster.executor.reset_mock() + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=NoHostAvailable('no host', {})): + self.control_connection._reconnect() + + assert self.control_connection._reconnection_handler is None + assert self.control_connection._reconnect_pending + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_reconnect_retries_when_contact_points_cannot_resolve(self): + # A DNS failure leaves no connection that can trigger a heartbeat, so + # it must enter the normal reconnection backoff on its own. + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=UnresolvableContactPoints({})): + self.control_connection._reconnect() + + assert self.control_connection._reconnection_handler is not None + + def test_connect_does_not_raise_when_shutdown_beats_it(self): + # shutdown() ran while _reconnect_internal() was connecting, so + # _set_new_connection() declined to install anything. + self.control_connection._connection = None + self.cluster.protocol_version = 4 + + def shut_down_and_connect(): + self.control_connection._is_shutdown = True + return Mock() + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=shut_down_and_connect): + self.control_connection.connect() + + assert self.control_connection._connection is None + + def test_reconnect_does_not_park_a_handler_after_another_attempt_won(self): + # Two attempts can overlap: reconnect() only collapses ones that have + # not started. If this one fails after the other installed a live + # connection, a handler parked here would block every later + # reconnect() for the whole backoff and then replace that connection. + def lose_the_race(): + self.control_connection._connection_generation += 1 + raise NoHostAvailable('no host', {}) + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=lose_the_race): + self.control_connection._reconnect() + + assert self.control_connection._reconnection_handler is None + + def test_reconnect_parks_a_handler_when_no_attempt_won(self): + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=NoHostAvailable('no host', {})): + self.control_connection._reconnect() + + assert self.control_connection._reconnection_handler is not None + + def test_reconnection_handler_closes_the_connection_it_cannot_hand_off(self): + # The ControlConnection was collected while the handler was backing + # off. run() has already marked the connection as handed off, so + # nothing else is left to close it. + handler = self._make_reconnection_handler() + owner = Mock() + handler.control_connection = weakref.proxy(owner) + del owner + conn = Mock() + + with patch.object(handler, 'try_reconnect', return_value=conn): + handler.run() + + conn.close.assert_called_once_with() + + def test_reconnect_defers_to_a_handler_left_by_a_failed_attempt(self): + self.cluster.executor.reset_mock() + self.control_connection.reconnect() + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=NoHostAvailable('no host', {})): + self.control_connection._reconnect() + + handler = self.control_connection._reconnection_handler + assert handler is not None + + self.control_connection.reconnect() + + # The handler installed by the failed attempt is retrying on its own + # schedule; starting another attempt would cancel it and restart that + # schedule from its initial delay. + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + assert self.control_connection._reconnection_handler is handler + assert not handler._cancelled + + def test_reconnect_is_queued_again_after_a_handler_fails_to_start(self): + # A handler that never got scheduled retries nothing, so it must not + # be left in the slot for reconnect() to defer to forever. + self.cluster.scheduler.schedule.side_effect = RuntimeError( + 'scheduler is shut down') + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=NoHostAvailable('no host', {})): + with self.assertRaises(RuntimeError): + self.control_connection._reconnect() + + assert self.control_connection._reconnection_handler is None + + self.cluster.executor.reset_mock() + self.control_connection.reconnect() + + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_returning_a_defunct_connection_does_not_restart_the_backoff(self): + # The heartbeat hands a defunct control connection back once per + # idle_heartbeat_interval for as long as it stays defunct. Each of + # those must leave the parked handler's schedule alone. + self.cluster.executor.reset_mock() + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=NoHostAvailable('no host', {})): + self.control_connection._reconnect() + + handler = self.control_connection._reconnection_handler + assert handler is not None + self.cluster.executor.reset_mock() + + self.connection.is_defunct = True + for _ in range(3): + self.control_connection.return_connection(self.connection) + + self.cluster.executor.submit.assert_not_called() + assert self.control_connection._reconnection_handler is handler + assert not handler._cancelled + + def test_heartbeat_does_not_restart_an_exhausted_finite_schedule(self): + self.cluster.reconnection_policy = ConstantReconnectionPolicy( + 0, max_attempts=1) + self.connection.is_defunct = True + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=NoHostAvailable('no host', {})): + self.control_connection._reconnect() + handler = self.control_connection._reconnection_handler + assert handler is not None + + # Run the only scheduled retry. Its failure exhausts the finite + # schedule and releases the handler slot. + handler.run() + + assert self.control_connection._reconnection_handler is None + self.cluster.executor.reset_mock() + self.connection.is_defunct = True + + # ConnectionHeartbeat returns this same defunct connection on every + # interval; none of those passes may create a new retry schedule. + for _ in range(3): + self.control_connection.return_connection(self.connection) + + self.cluster.executor.submit.assert_not_called() + + def test_final_handler_attempt_preserves_heartbeat_trigger(self): + self.cluster.reconnection_policy = ConstantReconnectionPolicy( + 0, max_attempts=1) + self.connection.is_defunct = True + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=NoHostAvailable('no host', {})): + self.control_connection._reconnect() + + handler = self.control_connection._reconnection_handler + assert handler is not None + self.cluster.executor.reset_mock() + + def fail_after_heartbeat(): + self.control_connection.return_connection(self.connection) + raise NoHostAvailable('no host', {}) + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=fail_after_heartbeat): + # The only scheduled retry receives a heartbeat trigger while it + # is running and then exhausts the finite schedule. + handler.run() + + assert self.control_connection._reconnection_handler is None + assert self.control_connection._reconnection_exhausted_connection \ + is None + assert self.control_connection._reconnect_pending + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_nonfinal_handler_attempt_consumes_heartbeat_trigger(self): + self.cluster.reconnection_policy = ConstantReconnectionPolicy( + 0, max_attempts=2) + self.connection.is_defunct = True + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=NoHostAvailable('no host', {})): + self.control_connection._reconnect() + + handler = self.control_connection._reconnection_handler + assert handler is not None + self.cluster.executor.reset_mock() + + def fail_after_heartbeat(): + self.control_connection.return_connection(self.connection) + raise NoHostAvailable('no host', {}) + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=fail_after_heartbeat): + # This retry still has another scheduled attempt to own the + # heartbeat trigger, so it must not start a new schedule. + handler.run() + + assert self.control_connection._reconnection_handler is handler + self.cluster.executor.submit.assert_not_called() + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=NoHostAvailable('no host', {})): + handler.run() + + assert self.control_connection._reconnection_handler is None + assert self.control_connection._reconnection_exhausted_connection \ + is self.connection + self.cluster.executor.submit.assert_not_called() + + def test_heartbeat_does_not_restart_an_empty_reconnection_schedule(self): + self.cluster.reconnection_policy = ExponentialReconnectionPolicy( + 1.0, 2.0, max_attempts=0) + self.connection.is_defunct = True + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=NoHostAvailable('no host', {})): + self.control_connection._reconnect() + + assert self.control_connection._reconnection_handler is None + assert self.control_connection._reconnection_exhausted_connection \ + is self.connection + self.cluster.executor.reset_mock() + + # No retries means that recurring heartbeat returns must not turn the + # empty schedule into one fresh immediate attempt per interval. + for _ in range(3): + self.control_connection.return_connection(self.connection) + + self.cluster.executor.submit.assert_not_called() + + def test_exhausted_replacement_does_not_suppress_a_later_failure(self): + self.cluster.reconnection_policy = ConstantReconnectionPolicy( + 0, max_attempts=1) + host = self.cluster.metadata.get_host_by_host_id('uuid1') + + # Removing the connected host starts a proactive replacement while + # the existing control connection is still healthy. + self.control_connection.on_remove(host) + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=NoHostAvailable('no host', {})): + self.control_connection._reconnect() + handler = self.control_connection._reconnection_handler + assert handler is not None + handler.run() + + assert self.control_connection._reconnection_handler is None + assert self.control_connection._reconnection_exhausted_connection \ + is None + + # A later, independent failure of the old connection must get a new + # retry schedule rather than being mistaken for the exhausted one. + self.cluster.executor.reset_mock() + self.connection.is_defunct = True + self.control_connection.return_connection(self.connection) + + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_failure_between_proactive_retries_starts_a_fresh_schedule(self): + self.cluster.reconnection_policy = ConstantReconnectionPolicy( + 0, max_attempts=1) + host = self.cluster.metadata.get_host_by_host_id('uuid1') + + # Start a proactive replacement while the existing control connection + # is healthy, then park the handler between its scheduled attempts. + self.control_connection.on_remove(host) + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=NoHostAvailable('no host', {})): + self.control_connection._reconnect() + + handler = self.control_connection._reconnection_handler + assert handler is not None + assert handler._failed_connection is None + self.cluster.executor.reset_mock() + + # With heartbeats disabled, this is the only failure notification the + # connection supplies. The proactive handler must retain it until its + # own finite schedule exhausts, then start a failure-owned schedule. + self.connection.is_defunct = True + self.control_connection.return_connection(self.connection) + self.cluster.executor.submit.assert_not_called() + + with patch.object(self.control_connection, '_reconnect_internal', + side_effect=NoHostAvailable('no host', {})): + handler.run() + + assert self.control_connection._reconnection_handler is None + assert self.control_connection._reconnect_pending + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_reconnect_is_queued_again_after_a_rejected_submission(self): + self.cluster.executor.reset_mock() + self.cluster.is_shutdown = True + self.addCleanup(setattr, self.cluster, 'is_shutdown', False) + + self.control_connection.reconnect() + + self.cluster.executor.submit.assert_not_called() + + self.cluster.is_shutdown = False + self.control_connection.reconnect() + + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_reconnect_is_queued_again_after_a_raising_submission(self): + self.cluster.executor.reset_mock() + self.cluster.executor.submit.side_effect = RuntimeError( + 'cannot schedule new futures after shutdown') + + with self.assertRaises(RuntimeError): + self.control_connection.reconnect() + + self.cluster.executor.submit.side_effect = None + self.cluster.executor.reset_mock() + + self.control_connection.reconnect() + + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + def test_defunct_control_reconnects_when_down_dispatch_is_dropped(self): # on_down() marks the host down but the executor refuses the DOWN # callback, so nothing else will reconnect the control connection. @@ -653,51 +1261,59 @@ def test_refresh_network_local_preserves_known_unix_endpoint(self): assert host_index[local_host] is not None assert Cluster.get_control_connection_host(self.cluster) is local_host - connection_error = ConnectionException('control connection failed') + # A DOWN transition discounted because a usable session pool remains + # queues no control on_down callback, so the reconnect is direct. + self._discount_down_for(local_host) self.connection.is_defunct = True - self.connection.last_error = connection_error - # Model a conviction whose DOWN transition is discounted because a - # usable session pool remains: no control on_down callback is queued. - self.cluster.signal_connection_failure = Mock(return_value=False) + self.connection.last_error = ConnectionException( + 'control connection failed') self.cluster.executor.reset_mock() self.control_connection._signal_error() - self.cluster.signal_connection_failure.assert_called_once_with( - local_host, connection_error, is_host_addition=False) + assert local_host.is_up is True + self.cluster.on_down_potentially_blocking.assert_not_called() self.cluster.executor.submit.assert_called_once_with( self.control_connection._reconnect) def test_unix_signal_error_reconnects_if_down_notification_suppressed(self): _, local_host = self._discover_local_host_over_unix() - connection_error = ConnectionException('control connection failed') + session = self._discount_down_for(local_host) self.connection.is_defunct = True - self.connection.last_error = connection_error - self.cluster.signal_connection_failure = Mock(return_value=False) + self.connection.last_error = ConnectionException( + 'control connection failed') self.cluster.executor.reset_mock() self.control_connection._signal_error() - self.cluster.signal_connection_failure.assert_called_once_with( - local_host, connection_error, is_host_addition=False) + # The discount is what suppresses the notification here, which means + # the host really was resolved from the Unix endpoint: an unresolved + # host would reconnect without ever consulting a session pool. + session.get_pool_state.assert_called_once_with() + assert local_host.is_up is True + self.cluster.on_down_potentially_blocking.assert_not_called() self.cluster.executor.submit.assert_called_once_with( self.control_connection._reconnect) def test_tcp_route_mismatch_reconnects_if_down_notification_suppressed(self): self.control_connection.refresh_node_list_and_token_map() local_host = self.cluster.metadata.get_host_by_host_id('uuid1') + local_host.set_up() self.connection.endpoint = DefaultEndPoint('192.168.1.0', 19042) self.connection.original_endpoint = local_host.endpoint - connection_error = ConnectionException('control connection failed') + session = self._discount_down_for(local_host) self.connection.is_defunct = True - self.connection.last_error = connection_error - self.cluster.signal_connection_failure = Mock(return_value=False) + self.connection.last_error = ConnectionException( + 'control connection failed') self.cluster.executor.reset_mock() self.control_connection._signal_error() - self.cluster.signal_connection_failure.assert_called_once_with( - local_host, connection_error, is_host_addition=False) + # As above: reaching the discount proves the host was resolved over + # the original endpoint despite the connection's route mismatch. + session.get_pool_state.assert_called_once_with() + assert local_host.is_up is True + self.cluster.on_down_potentially_blocking.assert_not_called() self.cluster.executor.submit.assert_called_once_with( self.control_connection._reconnect) @@ -733,16 +1349,15 @@ def test_route_mismatch_signal_error_reconnects_if_host_already_down(self): _, local_host = self._discover_local_host_over_unix() self._refresh_control_connection_over_network() local_host.set_down() - connection_error = ConnectionException('control connection failed') + self._use_cluster_down_handling() self.connection.is_defunct = True - self.connection.last_error = connection_error - self.cluster.signal_connection_failure = Mock(return_value=False) + self.connection.last_error = ConnectionException( + 'control connection failed') self.cluster.executor.reset_mock() self.control_connection._signal_error() - self.cluster.signal_connection_failure.assert_called_once_with( - local_host, connection_error, is_host_addition=False) + self.cluster.on_down_potentially_blocking.assert_not_called() self.cluster.executor.submit.assert_called_once_with( self.control_connection._reconnect) @@ -750,17 +1365,16 @@ def test_route_mismatch_signal_error_reconnects_if_host_reconnecting(self): _, local_host = self._discover_local_host_over_unix() self._refresh_control_connection_over_network() local_host.get_and_set_reconnection_handler(Mock()) - connection_error = ConnectionException('control connection failed') + self._use_cluster_down_handling() self.connection.is_defunct = True - self.connection.last_error = connection_error - - self.cluster.signal_connection_failure = Mock(return_value=False) + self.connection.last_error = ConnectionException( + 'control connection failed') self.cluster.executor.reset_mock() self.control_connection._signal_error() - self.cluster.signal_connection_failure.assert_called_once_with( - local_host, connection_error, is_host_addition=False) + assert local_host.is_up is False + self.cluster.on_down_potentially_blocking.assert_not_called() self.cluster.executor.submit.assert_called_once_with( self.control_connection._reconnect) From e843bca128cd2741c95fcf8ddd67ecb25ac75ad8 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Mon, 21 Sep 2026 21:50:44 -0400 Subject: [PATCH 3/4] changelog: document control-connection recovery --- CHANGELOG.rst | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index bd2f2b28be..98e1b1bac6 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -29,6 +29,40 @@ Features statements skip re-sending result metadata on EXECUTE, and the driver automatically refreshes cached metadata when the server detects a schema change (DRIVER-153) +Bug Fixes +--------- +* A defunct control connection is no longer left unreconnected when the cluster does not + run DOWN handling for its host (#847). ``ControlConnection._signal_error()`` treated the + conviction policy accepting a failure as a guarantee that a DOWN callback would + reconnect it, but the two are not the same: ``Cluster.on_down()`` deliberately skips + DOWN handling when a session pool to the host is still open, when the host is already + down or already reconnecting, and when pool creation is disabled -- and the default + ``SimpleConvictionPolicy`` rejects the conviction outright for ``OperationTimedOut``. + In all of those cases the control connection stayed defunct with nothing scheduled to + replace it. ``Cluster.on_down()`` and ``Cluster.signal_connection_failure()`` now return + whether DOWN handling was actually dispatched, and the control connection reconnects + directly whenever it was not. This applies uniformly to TCP, Unix socket, + alternate-route and stable host-ID connections, and does not duplicate the reconnect + that an accepted DOWN transition already performs. Two related cases are fixed with + it: a control connection whose reconnection attempts are already backing off no longer + has that schedule cancelled and restarted from its initial delay by every further + error, and a reconnection handler that has stopped for good -- once its retry schedule + is exhausted -- now releases the slot it occupies, so a later error starts a fresh + reconnection instead of mistaking the dead handler for one still retrying. The slot is + likewise released once a connection is installed, so a handler whose backoff outlived + the reconnection that succeeded without it no longer blocks the next one. Every release + is identity-checked, so a handler that stops can only ever clear itself: it cannot evict + a replacement another thread installed while it was handing its connection over, which + would have left that replacement retrying where nothing could find it. An attempt that + fails while an overlapping one succeeds no longer parks a handler in the slot the + successful attempt emptied, which would have blocked reconnection for the length of its + backoff and then replaced a healthy control connection. +* The control connection is no longer closed immediately after a reconnection handler + restores it. ``_ReconnectionHandler.run()`` closed the connection it had just opened, + which is right for the host handler that only uses it to probe the host, but left the + control connection dead the moment its backoff finally succeeded -- until a heartbeat + noticed, or forever with ``idle_heartbeat_interval=0``. + Others ------ * ``DCAwareRoundRobinPolicy.local_dc`` is now read-only. It is set by the constructor, From 284196047c08add85e06785aa52f14d2bdc2caf4 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Tue, 22 Sep 2026 12:03:13 -0400 Subject: [PATCH 4/4] control-connection: preserve DOWN recovery callbacks Isolate session and listener failures so asynchronous DOWN handling still starts host reconnection after reporting successful dispatch. Keep heartbeat submission outside the reconnection lock and use the control-handler type invariant directly.\n\nDocument policy-driven retry cadence and finite-schedule exhaustion. --- CHANGELOG.rst | 7 ++++++- cassandra/cluster.py | 16 +++++++++++----- tests/unit/test_cluster.py | 32 ++++++++++++++++++++++++++++++++ 3 files changed, 49 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 98e1b1bac6..b7389b2552 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -56,7 +56,12 @@ Bug Fixes would have left that replacement retrying where nothing could find it. An attempt that fails while an overlapping one succeeds no longer parks a handler in the slot the successful attempt emptied, which would have blocked reconnection for the length of its - backoff and then replaced a healthy control connection. + backoff and then replaced a healthy control connection. Once retry handling begins, its + cadence follows ``reconnection_policy`` up to ``max_reconnection_delay`` instead of + being restarted by each ``idle_heartbeat_interval``. Operators that need a shorter + recovery bound should configure a lower maximum delay. A finite schedule gives up after + its configured attempts, and recurring heartbeat returns for the same failed connection + do not re-arm it; continued automatic recovery requires an infinite schedule. * The control connection is no longer closed immediately after a reconnection handler restores it. ``_ReconnectionHandler.run()`` closed the connection it had just opened, which is right for the host handler that only uses it to probe the host, but left the diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 0df2f1621f..1cf0aff4a7 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -2038,10 +2038,16 @@ def on_down_potentially_blocking(self, host, is_host_addition): except Exception: log.exception("Error in control connection down handler for host %s", host) for session in tuple(self.sessions): - session.on_down(host) + try: + session.on_down(host) + except Exception: + log.exception("Error marking host %s down in session", host) for listener in self.listeners: - listener.on_down(host) + try: + listener.on_down(host) + except Exception: + log.exception("Error in host state listener down handler for host %s", host) self._start_reconnector(host, is_host_addition) @@ -4198,7 +4204,7 @@ def reconnect(self): with self._reconnection_lock: if self._reconnection_handler is not None: handler = self._reconnection_handler - if getattr(handler, '_is_running', False) is True: + if handler._is_running: handler._reconnect_requested = True # A handler can have been started proactively while the @@ -4209,7 +4215,7 @@ def reconnect(self): # heartbeats for an already-failed connection do not extend a # finite schedule indefinitely. connection = self._connection - if (getattr(handler, '_failed_connection', None) is None and + if (handler._failed_connection is None and connection is not None and (connection.is_defunct or connection.is_closed)): handler._failed_connection = connection @@ -4961,7 +4967,7 @@ def return_connection(self, connection): log.debug("[control connection] Reconnection schedule is " "exhausted for the defunct connection") return - self.reconnect() + self.reconnect() def _stop_scheduler(scheduler, thread): diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 36302ba030..c70e7ed921 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -822,6 +822,38 @@ def test_down_handling_continues_when_the_control_connection_raises(self): self.cluster._start_reconnector.assert_called_once_with( self.host, False) + def test_down_handling_continues_when_a_session_raises(self): + del self.cluster.on_down_potentially_blocking # use the real method + failing_session = Mock() + failing_session.on_down.side_effect = RuntimeError( + 'session down handling failed') + following_session = Mock() + self.cluster.sessions = [failing_session, following_session] + self.cluster._start_reconnector = Mock() + body = Cluster.on_down_potentially_blocking.__wrapped__ + + body(self.cluster, self.host, is_host_addition=False) + + following_session.on_down.assert_called_once_with(self.host) + self.cluster._start_reconnector.assert_called_once_with( + self.host, False) + + def test_down_handling_continues_when_a_listener_raises(self): + del self.cluster.on_down_potentially_blocking # use the real method + failing_listener = Mock() + failing_listener.on_down.side_effect = RuntimeError( + 'listener down handling failed') + following_listener = Mock() + self.cluster._listeners = {failing_listener, following_listener} + self.cluster._start_reconnector = Mock() + body = Cluster.on_down_potentially_blocking.__wrapped__ + + body(self.cluster, self.host, is_host_addition=False) + + following_listener.on_down.assert_called_once_with(self.host) + self.cluster._start_reconnector.assert_called_once_with( + self.host, False) + def test_signal_connection_failure_rejected_by_conviction_policy(self): error = ConnectionException("connection failed") self.host.signal_connection_failure = Mock(return_value=False)