Skip to content

cluster: discard in-flight UP completion for retired hosts #1124

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dkropachev merged 5 commits into from
Oct 12, 2026
+608 −88
Merged

cluster: discard in-flight UP completion for retired hosts #1124

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
c52b3ea
cluster: discard UP completion for retired hosts
dkropachev Oct 10, 2026
3d84763
metadata: serialize host replacement with UP completion
dkropachev Oct 11, 2026
3dbb6c3
cluster: serialize peers refresh removal with UP completion
dkropachev Oct 10, 2026
e57670d
cluster: release callback lock before removal notifications
dkropachev Oct 12, 2026
bff868b
metadata: serialize endpoint moves with UP completion
dkropachev Oct 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
148 changes: 97 additions & 51 deletions cassandra/cluster.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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 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):
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):
"""
Expand All @@ -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

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -2861,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:
Expand All @@ -2898,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,
Expand Down Expand Up @@ -3090,11 +3126,18 @@ 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:
# 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:
Comment thread
dkropachev marked this conversation as resolved.
if (self._is_current_host(host) and
self.metadata.remove_host(host)):
self._retire_host(host)
log.info("Cassandra host %s removed", host)
else:
return
self._notify_host_removed(
host, trigger_reconciliation, suppress_callback_errors)

def register_listener(self, listener):
"""
Expand Down Expand Up @@ -6420,7 +6463,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:
Expand Down
98 changes: 61 additions & 37 deletions cassandra/metadata.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
from bisect import bisect_left
from collections import defaultdict
from collections.abc import Mapping
from contextlib import ExitStack, nullcontext
from enum import Enum
from functools import total_ordering
from hashlib import md5
Expand Down Expand Up @@ -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)
Expand All @@ -396,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:
Expand Down
Loading
Loading