From c52b3ea158c3c143659e76ea13ff5441999010cb Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Fri, 9 Oct 2026 23:26:17 -0400 Subject: [PATCH 1/5] cluster: discard UP completion for retired hosts An in-flight pool future can finish after its Host is removed and a same-ID replacement is registered. Serialize completion with removal and check object identity before marking the old host UP or cleaning up failed pool creation. Stop callbacks and pool reconciliation if a listener removes the host reentrantly. Cover stale success, failure, and exception results, synchronous paths, and removal while completion is in progress. Fixes: #1074 --- cassandra/cluster.py | 110 ++++++++++++------- tests/unit/test_cluster.py | 219 +++++++++++++++++++++++++++++++++++++ 2 files changed, 289 insertions(+), 40 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index fd28ef86cf..ba036de07f 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -2320,12 +2320,23 @@ def _session_register_user_types(self, session): session.user_type_registered(keyspace, udt_name, klass) def _cleanup_failed_on_up_handling(self, host): + if not self._is_current_host(host): + return self.profile_manager.on_down(host) + if not self._is_current_host(host): + return self.control_connection.on_down(host) for session in tuple(self.sessions): + if not self._is_current_host(host): + return session.remove_pool(host) - self._start_reconnector(host, is_host_addition=False) + if self._is_current_host(host): + self._start_reconnector(host, is_host_addition=False) + + def _is_current_host(self, host): + return (not host._is_removed and + self.metadata.get_host_by_host_id(host.host_id) is host) def _on_up_future_completed(self, host, futures, results, lock, finished_future): with lock: @@ -2339,33 +2350,43 @@ def _on_up_future_completed(self, host, futures, results, lock, finished_future) if futures: return - try: - # all futures have completed at this point - for exc in [f for f in results if isinstance(f, Exception)]: - log.error("Unexpected failure while marking node %s up:", host, exc_info=exc) - self._cleanup_failed_on_up_handling(host) - return + # Removal takes this lock before unpublishing the host, so a current + # completion cannot be overtaken between validation and callbacks. + with host._down_callbacks_lock: + try: + if not self._is_current_host(host): + return - if not all(results): - log.debug("Connection pool could not be created, not marking node %s up", host) - self._cleanup_failed_on_up_handling(host) - return + # all futures have completed at this point + for exc in [f for f in results if isinstance(f, Exception)]: + log.error("Unexpected failure while marking node %s up:", host, exc_info=exc) + self._cleanup_failed_on_up_handling(host) + return - log.info("Connection pools established for node %s", host) - # mark the host as up and notify all listeners - with host.lock: - host.set_up() - host._pending_host_addition = False - host._pending_host_addition_callback = None - for listener in self.listeners: - listener.on_up(host) - finally: - with host.lock: - host._currently_handling_node_up = False + if not all(results): + log.debug("Connection pool could not be created, not marking node %s up", host) + self._cleanup_failed_on_up_handling(host) + return - # see if there are any pools to add or remove now that the host is marked up - for session in tuple(self.sessions): - session.update_created_pools() + log.info("Connection pools established for node %s", host) + # mark the host as up and notify all listeners + with host.lock: + host.set_up() + host._pending_host_addition = False + host._pending_host_addition_callback = None + for listener in self.listeners: + if not self._is_current_host(host): + return + listener.on_up(host) + + # see if there are any pools to add or remove now that the host is marked up + for session in tuple(self.sessions): + if not self._is_current_host(host): + return + session.update_created_pools() + finally: + with host.lock: + host._currently_handling_node_up = False def on_up(self, host): """ @@ -2376,7 +2397,7 @@ def on_up(self, host): log.debug("Waiting to acquire lock for handling up status of node %s", host) with host.lock: - if host._is_removed: + if not self._is_current_host(host): log.debug("Ignoring up status for removed host %s", host) return @@ -2435,18 +2456,22 @@ def on_up(self, host): for future in futures: future.cancel() - self._cleanup_failed_on_up_handling(host) - - with host.lock: - host._currently_handling_node_up = False + with host._down_callbacks_lock: + try: + self._cleanup_failed_on_up_handling(host) + finally: + with host.lock: + host._currently_handling_node_up = False raise else: if not have_future: - with host.lock: - host.set_up() - host._pending_host_addition = False - host._pending_host_addition_callback = None - host._currently_handling_node_up = False + with host._down_callbacks_lock: + with host.lock: + if self._is_current_host(host): + host.set_up() + host._pending_host_addition = False + host._pending_host_addition_callback = None + host._currently_handling_node_up = False # for testing purposes return futures @@ -3090,11 +3115,16 @@ def remove_host(self, host, trigger_reconciliation=True, ``suppress_callback_errors`` keeps replacement cleanup moving if a removal listener raises. """ - if host and self.metadata.remove_host(host): - log.info("Cassandra host %s removed", host) - self.on_remove( - host, trigger_reconciliation=trigger_reconciliation, - suppress_callback_errors=suppress_callback_errors) + if host: + # Share the callback lock with UP and DOWN completion. Keep it + # through on_remove so completion sees a fully retired host. + with host._down_callbacks_lock: + if (self._is_current_host(host) and + self.metadata.remove_host(host)): + log.info("Cassandra host %s removed", host) + self.on_remove( + host, trigger_reconciliation=trigger_reconciliation, + suppress_callback_errors=suppress_callback_errors) def register_listener(self, listener): """ diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 21552e1441..8a244bbaba 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -413,6 +413,225 @@ def test_on_up_ignores_removed_host(self): listener.on_up.assert_not_called() assert not host._currently_handling_node_up + def test_retired_host_up_future_cannot_touch_replacement(self): + for outcome in ('success', 'failure', 'exception'): + with self.subTest(outcome=outcome): + cluster = Cluster() + self.addCleanup(cluster.shutdown) + cluster.profile_manager = Mock() + cluster.profile_manager.distance.return_value = HostDistance.LOCAL + cluster.control_connection.on_up = Mock() + cluster.control_connection.on_down = Mock() + cluster.control_connection.on_remove = Mock() + cluster._prepare_all_queries = Mock() + cluster._start_reconnector = Mock() + + host = Host('127.0.0.1', SimpleConvictionPolicy, + host_id=uuid.uuid4()) + cluster.metadata.add_or_return_host(host) + host.set_down() + future = Future() + session = new_session_with_pool_state() + session.add_or_renew_pool = Mock(return_value=future) + session.on_remove = Mock() + session.update_created_pools = Mock() + session.remove_pool = Mock(wraps=session.remove_pool) + cluster.sessions = (session,) + listener = Mock() + cluster.register_listener(listener) + + cluster.on_up(host) + cluster.remove_host(host) + replacement = Host('127.0.0.1', SimpleConvictionPolicy, + host_id=host.host_id) + assert replacement is not host + cluster.metadata.add_or_return_host(replacement) + replacement_pool = Mock(host=replacement, is_shutdown=False) + session._pools[replacement] = replacement_pool + remove_count = session.remove_pool.call_count + + if outcome == 'exception': + future.set_exception(RuntimeError('pool failed')) + else: + future.set_result(outcome == 'success') + + assert host.is_up is False + assert not host._currently_handling_node_up + assert cluster.metadata.get_host_by_host_id(host.host_id) is replacement + assert session._pools[replacement] is replacement_pool + assert session.remove_pool.call_count == remove_count + session.update_created_pools.assert_not_called() + replacement_pool.shutdown.assert_not_called() + listener.on_up.assert_not_called() + listener.on_down.assert_not_called() + cluster.profile_manager.on_down.assert_not_called() + cluster.control_connection.on_down.assert_not_called() + cluster._start_reconnector.assert_not_called() + + def test_retired_host_up_without_future_is_discarded(self): + cluster = Cluster() + self.addCleanup(cluster.shutdown) + cluster.profile_manager = Mock() + cluster.profile_manager.distance.return_value = HostDistance.LOCAL + cluster.control_connection.on_up = Mock() + cluster.control_connection.on_remove = Mock() + cluster._prepare_all_queries = Mock() + host = Host('127.0.0.1', SimpleConvictionPolicy, + host_id=uuid.uuid4()) + cluster.metadata.add_or_return_host(host) + host.set_down() + session = Mock() + session.add_or_renew_pool.side_effect = lambda *_args, **_kwargs: ( + cluster.remove_host(host)) + cluster.sessions = (session,) + + cluster.on_up(host) + + assert host.is_up is False + assert not host._currently_handling_node_up + + def test_retired_host_synchronous_up_failure_skips_cleanup(self): + cluster = Cluster() + self.addCleanup(cluster.shutdown) + cluster.profile_manager = Mock() + cluster.profile_manager.distance.return_value = HostDistance.LOCAL + cluster.control_connection.on_up = Mock() + cluster.control_connection.on_down = Mock() + cluster.control_connection.on_remove = Mock() + cluster._prepare_all_queries = Mock() + cluster._start_reconnector = Mock() + host = Host('127.0.0.1', SimpleConvictionPolicy, + host_id=uuid.uuid4()) + cluster.metadata.add_or_return_host(host) + host.set_down() + session = Mock() + + def remove_then_fail(*_args, **_kwargs): + cluster.remove_host(host) + raise RuntimeError('pool scheduling failed') + + session.add_or_renew_pool.side_effect = remove_then_fail + cluster.sessions = (session,) + + with self.assertLogs('cassandra.cluster', level='ERROR'): + with self.assertRaisesRegex(RuntimeError, 'pool scheduling failed'): + cluster.on_up(host) + + assert host.is_up is False + assert not host._currently_handling_node_up + cluster.profile_manager.on_down.assert_not_called() + cluster.control_connection.on_down.assert_not_called() + cluster._start_reconnector.assert_not_called() + + def test_remove_waits_for_inflight_up_completion(self): + cluster = Cluster() + self.addCleanup(cluster.shutdown) + cluster.profile_manager = Mock() + cluster.profile_manager.distance.return_value = HostDistance.LOCAL + cluster.control_connection.on_up = Mock() + cluster.control_connection.on_remove = Mock() + cluster._prepare_all_queries = Mock() + host = Host('127.0.0.1', SimpleConvictionPolicy, + host_id=uuid.uuid4()) + cluster.metadata.add_or_return_host(host) + host.set_down() + future = Future() + session = Mock() + session.add_or_renew_pool.return_value = future + cluster.sessions = (session,) + callback_started = Event() + release_callback = Event() + removal_attempted = Event() + removal_finished = Event() + callback_order = [] + + def on_listener_up(_host): + callback_started.set() + assert release_callback.wait(5) + callback_order.append('up') + + listener = Mock() + listener.on_up.side_effect = on_listener_up + listener.on_remove.side_effect = lambda _host: callback_order.append('remove') + cluster.register_listener(listener) + cluster.on_up(host) + + callback_lock = host._down_callbacks_lock + remove_thread = None + + class AttemptTrackingLock(object): + + def __enter__(self): + if threading.current_thread() is remove_thread: + removal_attempted.set() + callback_lock.acquire() + return self + + def __exit__(self, exc_type, exc_value, traceback): + callback_lock.release() + + host._down_callbacks_lock = AttemptTrackingLock() + completion_thread = Thread(target=lambda: future.set_result(True)) + + def remove_host(): + try: + cluster.remove_host(host) + finally: + removal_finished.set() + + remove_thread = Thread(target=remove_host) + completion_thread.start() + try: + assert callback_started.wait(5) + remove_thread.start() + assert removal_attempted.wait(5) + assert not removal_finished.is_set() + finally: + release_callback.set() + completion_thread.join(5) + if remove_thread.ident is not None: + remove_thread.join(5) + + assert not completion_thread.is_alive() + assert not remove_thread.is_alive() + assert callback_order.index('up') < callback_order.index('remove') + session.update_created_pools.assert_called_once_with() + assert host._is_removed + + def test_reentrant_removal_stops_up_completion(self): + cluster = Cluster() + self.addCleanup(cluster.shutdown) + cluster.profile_manager = Mock() + cluster.profile_manager.distance.return_value = HostDistance.LOCAL + cluster.control_connection.on_up = Mock() + cluster.control_connection.on_remove = Mock() + cluster._prepare_all_queries = Mock() + host = Host('127.0.0.1', SimpleConvictionPolicy, + host_id=uuid.uuid4()) + cluster.metadata.add_or_return_host(host) + host.set_down() + future = Future() + session = Mock() + session.add_or_renew_pool.return_value = future + cluster.sessions = (session,) + removing_listener = Mock() + removing_listener.on_up.side_effect = cluster.remove_host + following_listener = Mock() + cluster.register_listener(removing_listener) + cluster.register_listener(following_listener) + cluster._listeners = SimpleNamespace( + copy=lambda: (removing_listener, following_listener)) + cluster.on_up(host) + + future.set_result(True) + + assert host._is_removed + assert host.is_up is False + assert not host._currently_handling_node_up + session.update_created_pools.assert_not_called() + assert (removing_listener.on_up.call_count + + following_listener.on_up.call_count) == 1 + def test_remove_host_continues_after_policy_callback_failure(self): cluster = Cluster() self.addCleanup(cluster.shutdown) From 3d847630132a57062b87b734d3c84f3056cf517e Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Sat, 10 Oct 2026 20:16:11 -0400 Subject: [PATCH 2/5] metadata: serialize host replacement with UP completion A replacement could publish after an UP completion validated the old host but before it marked that host UP. Acquire the old host callback lock before publishing, and retry when the endpoint owner changes while acquiring it. Cover replacement racing an in-flight UP callback. Refs: #1074 --- cassandra/cluster.py | 4 +- cassandra/metadata.py | 37 ++++++++++++------ tests/unit/test_cluster.py | 78 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 106 insertions(+), 13 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index ba036de07f..e9a1e52bff 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -2350,8 +2350,8 @@ def _on_up_future_completed(self, host, futures, results, lock, finished_future) if futures: return - # Removal takes this lock before unpublishing the host, so a current - # completion cannot be overtaken between validation and callbacks. + # Removal and replacement hold this lock while unpublishing the host, + # so an UP completion cannot outlive its metadata identity. with host._down_callbacks_lock: try: if not self._is_current_host(host): diff --git a/cassandra/metadata.py b/cassandra/metadata.py index 2a790150b9..2ea32159b5 100644 --- a/cassandra/metadata.py +++ b/cassandra/metadata.py @@ -16,6 +16,7 @@ from bisect import bisect_left from collections import defaultdict from collections.abc import Mapping +from contextlib import nullcontext from enum import Enum from functools import total_ordering from hashlib import md5 @@ -362,17 +363,31 @@ def _add_or_replace_host(self, host, allow_replace=True): Returns ``(host, new, replaced_host)``. Callers notify listeners only after the replacement is visible in both host indexes. """ - with self._hosts_lock: - existing = self._hosts.get(host.host_id) - if existing is not None: - return existing, False, None - - old_id = self._host_id_by_endpoint.get(host.endpoint) - if old_id is not None and not allow_replace: - raise ValueError("Replacing an endpoint owner requires host addition signaling") - replaced_host = self._hosts.pop(old_id, None) - self._hosts[host.host_id] = host - self._host_id_by_endpoint[host.endpoint] = host.host_id + while True: + with self._hosts_lock: + old_id = self._host_id_by_endpoint.get(host.endpoint) + expected_owner = self._hosts.get(old_id) + + # UP completion acquires the old host lock before looking up its + # metadata identity. Acquire locks in that order here too. + owner_lock = (expected_owner._down_callbacks_lock if expected_owner + is not None else nullcontext()) + with owner_lock: + with self._hosts_lock: + old_id = self._host_id_by_endpoint.get(host.endpoint) + if self._hosts.get(old_id) is not expected_owner: + continue + + existing = self._hosts.get(host.host_id) + if existing is not None: + return existing, False, None + + if old_id is not None and not allow_replace: + raise ValueError("Replacing an endpoint owner requires host addition signaling") + replaced_host = self._hosts.pop(old_id, None) + self._hosts[host.host_id] = host + self._host_id_by_endpoint[host.endpoint] = host.host_id + break if replaced_host is not None: self._tablets.drop_tablets_by_host_id(replaced_host.host_id) diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 8a244bbaba..e8ee7cba1c 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -598,6 +598,84 @@ def remove_host(): session.update_created_pools.assert_called_once_with() assert host._is_removed + def test_replacement_waits_for_inflight_up_completion(self): + cluster = Cluster() + self.addCleanup(cluster.shutdown) + cluster.profile_manager = Mock() + cluster.profile_manager.distance.return_value = HostDistance.LOCAL + cluster.control_connection.on_up = Mock() + cluster.control_connection.on_remove = Mock() + cluster._prepare_all_queries = Mock() + cluster.on_add = Mock() + host = Host('127.0.0.1', SimpleConvictionPolicy, + host_id=uuid.uuid4()) + cluster.metadata.add_or_return_host(host) + host.set_down() + future = Future() + session = Mock() + session.add_or_renew_pool.return_value = future + cluster.sessions = (session,) + callback_started = Event() + release_callback = Event() + replacement_attempted = Event() + replacement_finished = Event() + callback_order = [] + + def on_listener_up(_host): + callback_started.set() + assert release_callback.wait(5) + callback_order.append('up') + + listener = Mock() + listener.on_up.side_effect = on_listener_up + listener.on_remove.side_effect = lambda _host: callback_order.append('remove') + cluster.register_listener(listener) + cluster.on_up(host) + + callback_lock = host._down_callbacks_lock + replacement_thread = None + + class AttemptTrackingLock(object): + + def __enter__(self): + if threading.current_thread() is replacement_thread: + replacement_attempted.set() + callback_lock.acquire() + return self + + def __exit__(self, exc_type, exc_value, traceback): + callback_lock.release() + + host._down_callbacks_lock = AttemptTrackingLock() + completion_thread = Thread(target=lambda: future.set_result(True)) + + def replace_host(): + try: + cluster.add_host(host.endpoint, host_id=uuid.uuid4()) + finally: + replacement_finished.set() + + replacement_thread = Thread(target=replace_host) + completion_thread.start() + try: + assert callback_started.wait(5) + replacement_thread.start() + assert replacement_attempted.wait(5) + assert cluster.metadata.get_host_by_host_id(host.host_id) is host + assert not replacement_finished.is_set() + finally: + release_callback.set() + completion_thread.join(5) + if replacement_thread.ident is not None: + replacement_thread.join(5) + + assert not completion_thread.is_alive() + assert not replacement_thread.is_alive() + assert replacement_finished.is_set() + assert callback_order.index('up') < callback_order.index('remove') + assert cluster.metadata.get_host_by_host_id(host.host_id) is None + assert cluster.metadata.get_host(host.endpoint).host_id != host.host_id + def test_reentrant_removal_stops_up_completion(self): cluster = Cluster() self.addCleanup(cluster.shutdown) From 3dbb6c3af37ebc0a48fdd579295ac5d2f01f30f5 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Sat, 10 Oct 2026 19:46:34 -0400 Subject: [PATCH 3/5] cluster: serialize peers refresh removal with UP completion Peers refresh removes unseen hosts directly from metadata. Hold the host callback lock and verify its identity before unpublishing it, so a running UP completion cannot mark a retired host UP after the refresh. Refs: #1074 --- cassandra/cluster.py | 5 +++- tests/unit/test_control_connection.py | 42 +++++++++++++++++++++++++++ 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index e9a1e52bff..8aa9b199d8 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -6450,7 +6450,10 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None, if old_host_id not in found_host_ids: should_rebuild_token_map = True log.debug("[control connection] Removing host not found in peers metadata: %r", old_host) - self._cluster.metadata.remove_host_by_host_id(old_host_id, old_host.endpoint) + # This metadata-only removal must not overtake UP completion. + with old_host._down_callbacks_lock: + if self._cluster.metadata.get_host_by_host_id(old_host_id) is old_host: + self._cluster.metadata.remove_host_by_host_id(old_host_id, old_host.endpoint) log.debug("[control connection] Finished fetching ring info") if partitioner and should_rebuild_token_map: diff --git a/tests/unit/test_control_connection.py b/tests/unit/test_control_connection.py index 7460894e56..4c76465c2b 100644 --- a/tests/unit/test_control_connection.py +++ b/tests/unit/test_control_connection.py @@ -2825,6 +2825,48 @@ def test_refresh_nodes_and_tokens_remove_host(self): assert 1 == len(self.cluster.metadata.removed_hosts) assert self.cluster.metadata.removed_hosts[0].address == "192.168.1.2" + def test_refresh_removal_waits_for_host_up_completion(self): + self.control_connection.refresh_node_list_and_token_map() + host = self.cluster.metadata.get_host_by_host_id(HOST_ID_3) + del self.connection.peer_results[1][1] + callback_lock = host._down_callbacks_lock + removal_attempted = Event() + refresh_finished = Event() + + class AttemptTrackingLock(object): + + def __enter__(self): + removal_attempted.set() + callback_lock.acquire() + return self + + def __exit__(self, exc_type, exc_value, traceback): + callback_lock.release() + + host._down_callbacks_lock = AttemptTrackingLock() + + def refresh(): + try: + self.control_connection.refresh_node_list_and_token_map() + finally: + refresh_finished.set() + + refresh_thread = Thread(target=refresh) + callback_lock.acquire() + try: + refresh_thread.start() + assert removal_attempted.wait(5) + assert self.cluster.metadata.get_host_by_host_id(HOST_ID_3) is host + assert not refresh_finished.is_set() + finally: + callback_lock.release() + if refresh_thread.ident is not None: + refresh_thread.join(5) + + assert not refresh_thread.is_alive() + assert refresh_finished.is_set() + assert self.cluster.metadata.get_host_by_host_id(HOST_ID_3) is None + def test_refresh_nodes_and_tokens_timeout(self): def bad_wait_for_responses(*args, **kwargs): From e57670d79fee45ac0d25cacb2502be97acc41d5c Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Sun, 11 Oct 2026 23:29:18 -0400 Subject: [PATCH 4/5] cluster: release callback lock before removal notifications Keep metadata unpublishing and the removed-state transition under the host callback lock. Run listeners and pool reconciliation after releasing it so network refreshes and user callbacks cannot stall other host transitions. Keep on_remove's existing signature by moving notification dispatch to a private helper. Cover the lock boundary with a concurrent removal callback test. --- cassandra/cluster.py | 43 +++++++++++++++++++++++++------------- tests/unit/test_cluster.py | 33 +++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 15 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 8aa9b199d8..0e1370f089 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -2886,17 +2886,16 @@ def on_remove(self, host, trigger_reconciliation=True, if self.is_shutdown: return + self._retire_host(host) + self._notify_host_removed(host, trigger_reconciliation, + suppress_callback_errors) + + def _notify_host_removed(self, host, trigger_reconciliation, + suppress_callback_errors): + if self.is_shutdown: + return + log.debug("[cluster] Removing host %s", host) - # A Host object is never re-added after lifecycle removal; a later - # discovery creates a new object, even when it carries the same ID. - # Serialize the terminal transition with DOWN callbacks that already - # validated this Host, then mark it before removing its pools. - with host._down_callbacks_lock: - with host.lock: - host._is_removed = True - host.set_down() - host._pending_host_addition = False - host._pending_host_addition_callback = None def notify(callback, *args, **kwargs): if not suppress_callback_errors: @@ -2923,6 +2922,18 @@ def notify(callback, *args, **kwargs): if reconnection_handler: notify(reconnection_handler.cancel) + @staticmethod + def _retire_host(host): + # A Host object is never re-added after lifecycle removal; a later + # discovery creates a new object, even when it carries the same ID. + # Serialize this transition with DOWN callbacks before pool removal. + with host._down_callbacks_lock: + with host.lock: + host._is_removed = True + host.set_down() + host._pending_host_addition = False + host._pending_host_addition_callback = None + def signal_connection_failure(self, host, connection_exc, is_host_addition, expect_host_to_be_down=False, @@ -3116,15 +3127,17 @@ def remove_host(self, host, trigger_reconciliation=True, removal listener raises. """ if host: - # Share the callback lock with UP and DOWN completion. Keep it - # through on_remove so completion sees a fully retired host. + # Unpublish and retire the host under the callback lock so UP and + # DOWN completion see a terminal state. Run callbacks after release. with host._down_callbacks_lock: if (self._is_current_host(host) and self.metadata.remove_host(host)): + self._retire_host(host) log.info("Cassandra host %s removed", host) - self.on_remove( - host, trigger_reconciliation=trigger_reconciliation, - suppress_callback_errors=suppress_callback_errors) + else: + return + self._notify_host_removed( + host, trigger_reconciliation, suppress_callback_errors) def register_listener(self, listener): """ diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index e8ee7cba1c..2d3f6c1e47 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -598,6 +598,39 @@ def remove_host(): session.update_created_pools.assert_called_once_with() assert host._is_removed + def test_remove_releases_callback_lock_before_notifications(self): + cluster = Cluster() + self.addCleanup(cluster.shutdown) + cluster.control_connection.on_remove = Mock() + host = Host('127.0.0.1', SimpleConvictionPolicy, + host_id=uuid.uuid4()) + cluster.metadata.add_or_return_host(host) + lock_acquired = Event() + + def on_listener_remove(_host): + assert cluster.metadata.get_host_by_host_id(host.host_id) is None + assert host._is_removed + + def acquire_callback_lock(): + with host._down_callbacks_lock: + lock_acquired.set() + + lock_thread = Thread(target=acquire_callback_lock) + lock_thread.start() + assert lock_acquired.wait(2) + lock_thread.join(2) + assert not lock_thread.is_alive() + + listener = Mock() + listener.on_remove.side_effect = on_listener_remove + cluster.register_listener(listener) + + cluster.remove_host(host) + + listener.on_remove.assert_called_once_with(host) + cluster.control_connection.on_remove.assert_called_once_with( + host, trigger_reconciliation=True) + def test_replacement_waits_for_inflight_up_completion(self): cluster = Cluster() self.addCleanup(cluster.shutdown) From bff868bb78351e5899f0bb57412dac867fcbd1c1 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Sun, 11 Oct 2026 23:41:27 -0400 Subject: [PATCH 5/5] metadata: serialize endpoint moves with UP completion Acquire source host callback locks before publishing endpoint replacements so in-flight UP completion cannot act on a retired host after its identity check. Take batch locks in stable order and release them before lifecycle callbacks. Cover the move/completion race with a deterministic test. --- cassandra/metadata.py | 63 +++++++++++++++++------------- tests/unit/test_cluster.py | 78 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 114 insertions(+), 27 deletions(-) diff --git a/cassandra/metadata.py b/cassandra/metadata.py index 2ea32159b5..d1c4ef8f9a 100644 --- a/cassandra/metadata.py +++ b/cassandra/metadata.py @@ -16,7 +16,7 @@ from bisect import bisect_left from collections import defaultdict from collections.abc import Mapping -from contextlib import nullcontext +from contextlib import ExitStack, nullcontext from enum import Enum from functools import total_ordering from hashlib import md5 @@ -411,32 +411,41 @@ def replace_host_endpoints(self, replacements): replacements = list(replacements) source_ids = set() destinations = set() - with self._hosts_lock: - for old_host, replacement in replacements: - if old_host.host_id != replacement.host_id: - raise ValueError("Endpoint replacement must preserve host_id") - if old_host.endpoint == replacement.endpoint: - raise ValueError("Endpoint replacement requires a new endpoint") - if old_host.host_id in source_ids: - raise ValueError("Host appears more than once in endpoint replacement") - if replacement.endpoint in destinations: - raise ValueError("Endpoint appears more than once in endpoint replacement") - if self._hosts.get(old_host.host_id) is not old_host: - raise ValueError("Host is no longer current in metadata") - source_ids.add(old_host.host_id) - destinations.add(replacement.endpoint) - - for old_host, replacement in replacements: - owner_id = self._host_id_by_endpoint.get(replacement.endpoint) - if owner_id is not None and owner_id not in source_ids: - raise ValueError("New endpoint belongs to another host") - - for old_host, _ in replacements: - if self._host_id_by_endpoint.get(old_host.endpoint) == old_host.host_id: - self._host_id_by_endpoint.pop(old_host.endpoint) - for _, replacement in replacements: - self._hosts[replacement.host_id] = replacement - self._host_id_by_endpoint[replacement.endpoint] = replacement.host_id + # UP completion takes a source host's callback lock before checking + # metadata identity. Take every source lock in a stable order before + # publishing the batch, then release them before lifecycle callbacks. + with ExitStack() as locks: + for old_host, _ in sorted( + replacements, + key=lambda pair: (str(pair[0].host_id), id(pair[0]))): + locks.enter_context(old_host._down_callbacks_lock) + with self._hosts_lock: + for old_host, replacement in replacements: + if old_host.host_id != replacement.host_id: + raise ValueError("Endpoint replacement must preserve host_id") + if old_host.endpoint == replacement.endpoint: + raise ValueError("Endpoint replacement requires a new endpoint") + if old_host.host_id in source_ids: + raise ValueError("Host appears more than once in endpoint replacement") + if replacement.endpoint in destinations: + raise ValueError( + "Endpoint appears more than once in endpoint replacement") + if self._hosts.get(old_host.host_id) is not old_host: + raise ValueError("Host is no longer current in metadata") + source_ids.add(old_host.host_id) + destinations.add(replacement.endpoint) + + for old_host, replacement in replacements: + owner_id = self._host_id_by_endpoint.get(replacement.endpoint) + if owner_id is not None and owner_id not in source_ids: + raise ValueError("New endpoint belongs to another host") + + for old_host, _ in replacements: + if self._host_id_by_endpoint.get(old_host.endpoint) == old_host.host_id: + self._host_id_by_endpoint.pop(old_host.endpoint) + for _, replacement in replacements: + self._hosts[replacement.host_id] = replacement + self._host_id_by_endpoint[replacement.endpoint] = replacement.host_id def remove_host(self, host): with self._hosts_lock: diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 2d3f6c1e47..de6cef302c 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -709,6 +709,84 @@ def replace_host(): assert cluster.metadata.get_host_by_host_id(host.host_id) is None assert cluster.metadata.get_host(host.endpoint).host_id != host.host_id + def test_endpoint_move_waits_for_inflight_up_completion(self): + cluster = Cluster() + self.addCleanup(cluster.shutdown) + cluster.profile_manager = Mock() + cluster.profile_manager.distance.return_value = HostDistance.LOCAL + cluster.control_connection.on_up = Mock() + cluster.control_connection.on_remove = Mock() + cluster._prepare_all_queries = Mock() + cluster.on_add = Mock() + host = Host(DefaultEndPoint('127.0.0.1'), SimpleConvictionPolicy, + host_id=uuid.uuid4()) + cluster.metadata.add_or_return_host(host) + host.set_down() + future = Future() + session = Mock() + session.add_or_renew_pool.return_value = future + cluster.sessions = (session,) + callback_started = Event() + release_callback = Event() + move_attempted = Event() + callback_order = [] + + def on_listener_up(_host): + callback_started.set() + assert release_callback.wait(5) + callback_order.append('up') + + listener = Mock() + listener.on_up.side_effect = on_listener_up + listener.on_remove.side_effect = lambda _host: callback_order.append('remove') + cluster.register_listener(listener) + cluster.on_up(host) + + callback_lock = host._down_callbacks_lock + move_thread = None + + class AttemptTrackingLock(object): + + def __enter__(self): + if threading.current_thread() is move_thread: + move_attempted.set() + callback_lock.acquire() + return self + + def __exit__(self, exc_type, exc_value, traceback): + callback_lock.release() + + host._down_callbacks_lock = AttemptTrackingLock() + completion_thread = Thread(target=lambda: future.set_result(True)) + move_errors = [] + + def move_host(): + try: + cluster.replace_host_endpoint( + host, DefaultEndPoint('127.0.0.2')) + except Exception as exc: + move_errors.append(exc) + + move_thread = Thread(target=move_host) + completion_thread.start() + try: + assert callback_started.wait(5) + move_thread.start() + assert move_attempted.wait(5) + assert cluster.metadata.get_host_by_host_id(host.host_id) is host + finally: + release_callback.set() + completion_thread.join(5) + if move_thread.ident is not None: + move_thread.join(5) + + assert not completion_thread.is_alive() + assert not move_thread.is_alive() + assert not move_errors + assert callback_order.index('up') < callback_order.index('remove') + assert host._is_removed + assert cluster.metadata.get_host_by_host_id(host.host_id) is not host + def test_reentrant_removal_stops_up_completion(self): cluster = Cluster() self.addCleanup(cluster.shutdown)