diff --git a/cassandra/_lifecycle.py b/cassandra/_lifecycle.py new file mode 100644 index 0000000000..f5ae912777 --- /dev/null +++ b/cassandra/_lifecycle.py @@ -0,0 +1,542 @@ +# Copyright DataStax, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Internal host lifecycle coordination; not part of the driver public API. + +Classes use ordinary names within this private module. Import the module +explicitly for internal use; no names are exported by wildcard imports. +""" + +from collections import deque +from concurrent.futures import Future +from contextlib import contextmanager, nullcontext +from functools import partial, wraps +import logging +from threading import local, RLock +import weakref +from weakref import WeakValueDictionary + +from cassandra import DriverException + +__all__ = () + +# Preserve the existing logging category for cluster lifecycle events. +log = logging.getLogger("cassandra.cluster") + + +class EndpointMoveRejected(ValueError): + """A batch could not publish because its metadata snapshot went stale.""" + + +class PoolFailureReservation: + """Fence an admitted pool failure without locking across policy callbacks.""" + + def __init__(self, session, host, generation, expected_pool, intent, endpoint): + self.session = session + self.host = host + self.generation = generation + self.expected_pool = expected_pool + self.intent = intent + self.endpoint = endpoint + + def owns_locked(self): + return (self.session._pool_intents_locked().get(self.host) is self.intent and + self.session._owns_pool_generation_locked( + self.host, self.generation, self.expected_pool, self.endpoint)) + + def is_current(self): + with self.session._lock: + with self.host.lock: + return self.owns_locked() + + +class LifecycleCoordinator: + """Own host admission, readiness, retirement and notification ordering. + + Metadata owns indexes and token maps; Cluster and Session perform the + policy, connection and pool operations. This coordinator publishes their + transitions under a short lock and delivers FIFO notifications unlocked. + A callback may synchronously publish another topology. Its notifications + join the queue instead of recursively overtaking the current lifecycle. + """ + + def __init__(self, cluster=None): + self._cluster = weakref.proxy(cluster) if cluster is not None else None + self._lock = RLock() + self._local = local() + self._events = deque() + self._draining = False + self._requested = 0 + self._published = 0 + self._force_revisions = deque() + self._closed = False + self._retired = WeakValueDictionary() + + @property + def closed(self): + # Readable under host.lock without acquiring the publication lock. + return self._closed or (self._cluster is not None and + self._cluster.is_shutdown) + + @property + def delivering(self): + return getattr(self._local, 'delivering', False) + + def close(self): + """Reject new work and abandon queued work without joining callbacks.""" + with self._lock: + if self._closed: + return + self._closed = True + self._events.clear() + self._force_revisions.clear() + hosts = list(self._retired.values()) + if self._cluster is not None: + hosts.extend(self._cluster.metadata.all_hosts()) + self._retired.clear() + reconnectors = [] + for host in hosts: + with host.lock: + host._pending_location = None + host._pending_location_token = None + host._lifecycle_pending = False + host._pending_host_addition = False + host._pending_host_addition_callback = None + host._currently_handling_node_up = False + host._currently_handling_node_down = False + host._down_event_generation += 1 + handler = host.get_and_set_reconnection_handler(None) + if handler is not None: + reconnectors.append(handler) + for handler in reconnectors: + try: + handler.cancel() + except Exception: + log.exception("Error cancelling host reconnector during shutdown") + + def begin_up(self, host): + with host.lock: + if (not self.is_current(host) or host.is_up or + host._currently_handling_node_up): + return False + host._currently_handling_node_up = True + return True + + @staticmethod + def finish_up(host): + with host.lock: + host._currently_handling_node_up = False + + def begin_down(self, host, connected=False, expect_down=False, pool_failure=None): + """Return (generation, retry_only), or None if DOWN is discounted.""" + failure_lock = (pool_failure.session._lock + if pool_failure is not None else nullcontext()) + with failure_lock: + with host.lock: + if (self.closed or host._is_removed or connected or + (pool_failure is not None and not pool_failure.owns_locked())): + return None + was_up = host.is_up + host.set_down() + if (not was_up and (host.is_currently_reconnecting() or + host._currently_handling_node_down)): + return None + host._currently_handling_node_down = True + host._down_event_generation += 1 + return host._down_event_generation, not was_up and not expect_down + + def dispatch_down(self, host, is_host_addition, connected, expect_host_to_be_down, + on_add_reconnection, pool_failure=None): + # Publish DOWN state and FIFO admission together. Do not drain on + # this caller; the executor (or another drainer) owns delivery. + with self.transaction(drain=False): + admission = self.begin_down( + host, connected=connected, expect_down=expect_host_to_be_down, + pool_failure=pool_failure) + if admission is None: + return False + down_event_generation, restart_reconnector = admission + + if restart_reconnector: + if on_add_reconnection is None: + future = self._cluster._restart_reconnector( + host, is_host_addition, down_event_generation) + else: + future = self._cluster._restart_reconnector( + host, is_host_addition, down_event_generation, + on_add_reconnection=on_add_reconnection) + if future is None: + self.finish_down(host, down_event_generation) + # Restarting the host reconnector does not dispatch the DOWN + # callbacks that reconnect the control connection. + return False + + log.warning("Host %s has been marked down", host) + + if on_add_reconnection is None: + future = self._cluster.on_down_potentially_blocking( + host, is_host_addition, down_event_generation) + else: + future = self._cluster.on_down_potentially_blocking( + host, is_host_addition, down_event_generation, + on_add_reconnection=on_add_reconnection) + if future is None: + self.finish_down(host, down_event_generation) + return future is not None + + @staticmethod + def finish_down(host, generation): + with host.lock: + if generation == host._down_event_generation: + host._currently_handling_node_down = False + + def down_is_current(self, host, generation=None): + return (not self.closed and not host._is_removed and + (generation is None or generation == host._down_event_generation)) + + @staticmethod + def supersede_down(host): + """Invalidate queued recovery after terminal addition failure.""" + with host.lock: + host._down_event_generation += 1 + host._currently_handling_node_down = False + return host.get_and_set_reconnection_handler(None) + + @contextmanager + def transaction(self, drain=True): + try: + with self._lock: + self._local.depth = getattr(self._local, 'depth', 0) + 1 + try: + yield + finally: + self._local.depth -= 1 + finally: + if drain and not getattr(self._local, 'depth', 0): + self.drain() + + def reserve_refresh(self, force_token_rebuild=False): + with self._lock: + self._requested += 1 + if force_token_rebuild: + self._force_revisions.append(self._requested) + return self._requested + + def force_rebuild_required(self, revision): + """Keep force intent until this or a newer snapshot rebuilds tokens.""" + return bool(self._force_revisions and self._force_revisions[0] <= revision) + + def complete_token_rebuild(self, revision): + while self._force_revisions and self._force_revisions[0] <= revision: + self._force_revisions.popleft() + + def accept_refresh(self, revision): + # Called inside a publication transaction. + if self.closed or revision < self._published: + return False + self._published = revision + return True + + def submit(self, callback, *args, **kwargs): + """Queue a notification; delivery never has a synchronous result.""" + with self._lock: + if self.closed: + return + self._events.append(partial(callback, *args, **kwargs)) + if not getattr(self._local, 'depth', 0): + self.drain() + + def submit_async(self, executor, callback, *args, **kwargs): + """Queue before executor handoff so later notifications cannot overtake.""" + event = partial(callback, *args, **kwargs) + with self._lock: + if self.closed: + return None + self._events.append(event) + try: + return executor.submit(self.drain) + except Exception: + log.exception("Failed to schedule host lifecycle notifications") + # If no drainer has claimed this event, failed handoff cancels + # admission. Otherwise a concurrent submit already delivered it. + with self._lock: + try: + self._events.remove(event) + except ValueError: + if self.closed: + return None + else: + return None + completed = Future() + completed.set_result(None) + return completed + + def is_current(self, host): + """Check that this exact, nonterminal Host owns its metadata entry.""" + return (not self.closed and not host._is_removed and + self._cluster.metadata.get_host_by_host_id(host.host_id) is host) + + def dispatch_current(self, host, callback, *args, generation=None, **kwargs): + """Admit one callback; retirement may publish after admission.""" + with host.lock: + # DOWN work already owns a generation. A transient metadata + # reindex must not prevent its recovery notifications. + admitted = (self.is_current(host) if generation is None else + self.down_is_current(host, generation)) + if not admitted: + return False + callback(host, *args, **kwargs) + return True + + def submit_current(self, host, callback, *args, **kwargs): + """Check identity when queued work starts, not when it is submitted.""" + def deliver(): + if self.is_current(host): + callback(*args, **kwargs) + self.submit(deliver) + + @staticmethod + def reserve_addition(host, pending=True): + """Reserve a candidate before publication and pool reconciliation.""" + with host.lock: + host._lifecycle_pending = pending + + def begin_addition(self, host, on_reconnection=None): + with host.lock: + if not self.is_current(host): + return False + host._lifecycle_pending = True + self.remember_addition(host, on_reconnection) + return True + + @staticmethod + def remember_addition(host, on_reconnection=None): + """Retain addition intent across retries; callers hold host.lock.""" + host._pending_host_addition = True + if on_reconnection is not None: + host._pending_host_addition_callback = on_reconnection + + def release_addition(self, host): + """Let session reconciliation resume after an ordinary ADD failure.""" + with host.lock: + if self.is_current(host): + host._lifecycle_pending = False + + def mark_ready(self, host, set_up=True): + if not self.is_current(host): + return False + if set_up: + host.conviction_policy.reset() + with host.lock: + if not self.is_current(host): + return False + if set_up: + host.is_up = True + host._lifecycle_pending = False + host._pending_host_addition = False + host._pending_host_addition_callback = None + return True + + def _retire(self, 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. + # Pool/reconnection publication checks this terminal state under the + # same state lock. Notifications run later, without holding it. + with host.lock: + if host._is_removed: + return False + host._is_removed = True + self._retired[id(host)] = host + host.set_down() + host._pending_host_addition = False + host._pending_host_addition_callback = None + host._lifecycle_pending = False + host._pending_location = None + host._pending_location_token = None + host._currently_handling_node_up = False + host._currently_handling_node_down = False + host._down_event_generation += 1 + return True + + def retire(self, host, trigger_reconciliation=True, refresh_nodes=True): + with self.transaction(): + if not self._retire(host): + return + self.submit(self._cluster._notify_host_removed, host, + trigger_reconciliation, True, refresh_nodes) + + def remove(self, host, trigger_reconciliation=True, refresh_nodes=True): + with self.transaction(): + if (host is None or not self.is_current(host) or + not self._cluster.metadata.remove_host(host)): + return + self._retire(host) + log.info("Cassandra host %s removed", host) + self.submit(self._cluster._notify_host_removed, host, + trigger_reconciliation, True, refresh_nodes) + + def publish_host(self, candidate, signal, refresh_nodes, + reconcile_pools_on_failure, replace_existing_id): + """Publish discovery and queue displaced cleanup before new ADD.""" + with self.transaction(): + if self.closed: + raise DriverException("Cluster is already shut down") + self.reserve_addition(candidate, pending=signal) + host, new, displaced = self._cluster.metadata._add_or_replace_host( + candidate, allow_replace=signal) + if not new and host.endpoint != candidate.endpoint: + if not signal: + raise ValueError("Endpoint replacement requires host addition signaling") + if not replace_existing_id: + raise ValueError("Endpoint replacement is disabled for this host addition") + return self._cluster.replace_host_endpoint( + host, candidate.endpoint, candidate.datacenter, candidate.rack), False + self._notify_addition(host, new, displaced, signal, refresh_nodes, + reconcile_pools_on_failure) + return host, new + + def _notify_addition(self, host, new, displaced, signal, refresh_nodes, + reconcile_pools_on_failure): + """Order a displaced lifecycle's cleanup before its successor's ADD.""" + cluster = self._cluster + if displaced is not None: + log.info("Cassandra host %s removed", displaced) + try: + cluster.on_remove(displaced, trigger_reconciliation=False, + suppress_callback_errors=True) + except Exception: + log.exception("Error notifying removal of replaced host %s", + displaced) + reconcile_pools_on_failure = True + if new and signal: + self.reserve_addition(host) + log.info("New Cassandra host %r discovered", host) + cluster.on_add(host, refresh_nodes, + reconcile_pools_on_failure=reconcile_pools_on_failure) + + def publish_replacements(self, replacements, reconnect=None): + """Publish a batch and schedule retirement before replacement ADD.""" + with self.transaction(): + if self.closed: + raise DriverException("Cluster is already shut down") + for _, replacement in replacements: + self.reserve_addition(replacement) + try: + self._cluster.metadata.replace_host_endpoints(replacements) + except ValueError as exc: + raise EndpointMoveRejected(str(exc)) from exc + self._notify_replacements(replacements, reconnect) + + def _notify_replacements(self, replacements, reconnect): + cluster = self._cluster + for host, replacement in replacements: + log.info("Cassandra host %s moved from %s to %s", host.host_id, + host.endpoint, replacement.endpoint) + cluster.on_remove(host, trigger_reconciliation=False, + suppress_callback_errors=True) + try: + for _, replacement in replacements: + try: + cluster.on_add(replacement, refresh_nodes=False, + reconcile_pools_on_failure=True) + except Exception: + log.exception("Error notifying addition of moved host %s", + replacement) + with replacement.lock: + replacement._endpoint_add_retry_pending = True + cluster._schedule_failed_endpoint_add_retry(replacement) + finally: + if reconnect is not None: + self.submit(reconnect) + + def update_location(self, host, datacenter, rack): + with self.transaction(): + if self.closed: + return False + return self._queue_location(host, datacenter, rack) + + def _queue_location(self, host, datacenter, rack): + desired = (datacenter, rack) + pending = getattr(host, '_pending_location', None) + if (pending if pending is not None else + (host.datacenter, host.rack)) == desired: + return False + token = object() + host._pending_location = desired + host._pending_location_token = token + self.submit( + self._notify_location_changed, host, datacenter, rack, token) + return True + + def _notify_location_changed(self, host, datacenter, rack, token): + try: + if not self.is_current(host): + return + # Policies use the old DC/rack to remove membership. Publish the + # new location only after that callback, then invalidate replica + # caches by replacing their owner object, not clearing it in place. + self._cluster.profile_manager.on_down(host) + with self.transaction(): + if not self.is_current(host): + return + changed = self._cluster.metadata._update_host_location( + host, datacenter, rack) + if changed: + self.dispatch_current(host, self._cluster.profile_manager.on_up) + finally: + with self.transaction(): + if getattr(host, '_pending_location_token', None) is token: + host._pending_location = None + host._pending_location_token = None + + def drain(self): + with self._lock: + if self._draining or self.closed: + return + self._draining = True + try: + while True: + with self._lock: + if self.closed or not self._events: + self._events.clear() + self._draining = False + return + callback = self._events.popleft() + try: + self._local.delivering = True + callback() + except Exception: + log.exception("Error delivering host lifecycle notification") + finally: + self._local.delivering = False + except BaseException: + with self._lock: + self._draining = False + raise + + +def event(method): + @wraps(method) + def dispatch(self, *args, **kwargs): + return self._lifecycle.submit(method, self, *args, **kwargs) + dispatch._lifecycle_method = method + return dispatch + + +def transaction(method): + @wraps(method) + def publish(self, *args, **kwargs): + with self._lifecycle.transaction(): + return method(self, *args, **kwargs) + return publish diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 0e1370f089..30eb8a20a1 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -54,6 +54,7 @@ from cassandra import (ConsistencyLevel, AuthenticationFailed, OperationTimedOut, UnsupportedOperation, SchemaTargetType, DriverException, ProtocolVersion, UnresolvableContactPoints, DependencyException, InvalidRequest) +from cassandra import _lifecycle from cassandra.auth import _proxy_execute_key, PlainTextAuthProvider from cassandra.client_routes import ClientRoutesChangeType, ClientRoutesConfig, _ClientRoutesHandler from cassandra.connection import (ClientRoutesEndPointFactory, ConnectionException, ConnectionShutdown, @@ -158,10 +159,6 @@ def _connection_reduce_fn(val,import_fn): _NOT_SET = object() -class _EndpointMoveRejected(ValueError): - """A batch could not publish because its metadata snapshot went stale.""" - - def _cancel_and_notify(future): # A bare Future's cancel() doesn't wake concurrent.futures.wait() callers; # only set_running_or_notify_cancel() does (executors call it themselves). @@ -386,7 +383,14 @@ def new_f(self, *args, **kwargs): if self.is_shutdown: return try: - future = self.executor.submit(f, self, *args, **kwargs) + lifecycle_method = getattr(f, '_lifecycle_method', None) + if lifecycle_method is not None: + future = self._lifecycle.submit_async( + self.executor, lifecycle_method, self, *args, **kwargs) + if future is None: + return None + else: + future = self.executor.submit(f, self, *args, **kwargs) future.add_done_callback(_future_completed) return future except Exception: @@ -1788,6 +1792,7 @@ def __init__(self, and not client_routes_config.advanced_shard_awareness): self.shard_aware_options.disable_shardaware_port = True + self._lifecycle = _lifecycle.LifecycleCoordinator(self) self._listeners = set() self._listener_lock = Lock() @@ -2280,6 +2285,10 @@ def shutdown(self): def _shutdown(self, wait_for_executor): """Finish cleanup after shutdown has been claimed under ``self._lock``.""" + self._lifecycle.close() + if self._lifecycle.delivering: + wait_for_executor = False + if self._idle_heartbeat: self._idle_heartbeat.stop() @@ -2335,9 +2344,9 @@ def _cleanup_failed_on_up_handling(self, 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) + return self._lifecycle.is_current(host) + @_lifecycle.event def _on_up_future_completed(self, host, futures, results, lock, finished_future): with lock: futures.discard(finished_future) @@ -2350,44 +2359,39 @@ def _on_up_future_completed(self, host, futures, results, lock, finished_future) if futures: 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 + # Retirement can publish while a callback runs. Fence every remaining + # effect; the FIFO delivers removal after the current notification. + try: + if not self._is_current_host(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 + # 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 - 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 + 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 - 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) + log.info("Connection pools established for node %s", host) + if not self._lifecycle.mark_ready(host): + return + for listener in self.listeners: + if not self._lifecycle.dispatch_current(host, listener.on_up): + return - # 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 + # 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: + self._lifecycle.finish_up(host) + @_lifecycle.event def on_up(self, host): """ Intended for internal use only. @@ -2395,21 +2399,8 @@ def on_up(self, host): if self.is_shutdown or self.allow_control_connection_query_fallback == ControlConnectionQueryFallback.SkipPoolCreation: return - log.debug("Waiting to acquire lock for handling up status of node %s", host) - with host.lock: - if not self._is_current_host(host): - log.debug("Ignoring up status for removed host %s", host) - return - - if host._currently_handling_node_up: - log.debug("Another thread is already handling up status of node %s", host) - return - - if host.is_up: - log.debug("Host %s was already marked up", host) - return - - host._currently_handling_node_up = True + if not self._lifecycle.begin_up(host): + return log.debug("Starting to handle up status of node %s", host) have_future = False @@ -2422,24 +2413,34 @@ def on_up(self, host): log.debug("Now that host %s is up, cancelling the reconnection handler", host) reconnector.cancel() - if self.profile_manager.distance(host) != HostDistance.IGNORED: + if not self._is_current_host(host): + return + distance = self.profile_manager.distance(host) + if not self._is_current_host(host): + return + if distance != HostDistance.IGNORED: self._prepare_all_queries(host) log.debug("Done preparing all queries for host %s, ", host) for session in tuple(self.sessions): - session.remove_pool(host) + if not self._lifecycle.dispatch_current(host, session.remove_pool): + return log.debug("Signalling to load balancing policies that host %s is up", host) - self.profile_manager.on_up(host) + if not self._lifecycle.dispatch_current(host, self.profile_manager.on_up): + return log.debug("Signalling to control connection that host %s is up", host) - self.control_connection.on_up(host) + if not self._lifecycle.dispatch_current(host, self.control_connection.on_up): + return log.debug("Attempting to open new connection pools for host %s", host) futures_lock = Lock() futures_results = [] callback = partial(self._on_up_future_completed, host, futures, futures_results, futures_lock) for session in tuple(self.sessions): + if not self._is_current_host(host): + return future = session.add_or_renew_pool(host, is_host_addition=False) if future is not None: have_future = True @@ -2456,30 +2457,27 @@ def on_up(self, host): for future in futures: future.cancel() - with host._down_callbacks_lock: - try: - self._cleanup_failed_on_up_handling(host) - finally: - with host.lock: - host._currently_handling_node_up = False + try: + self._cleanup_failed_on_up_handling(host) + finally: + self._lifecycle.finish_up(host) raise else: if not have_future: - 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 + try: + self._lifecycle.mark_ready(host) + finally: + self._lifecycle.finish_up(host) def _start_reconnector(self, host, is_host_addition, - on_add_reconnection=None, start=True): + on_add_reconnection=None, start=True, + down_event_generation=None, restart=False): with host.lock: - if host._is_removed: + if not self._lifecycle.down_is_current(host, down_event_generation): + return None + if restart and (host.is_up is not False or + host._currently_handling_node_up or + host.is_currently_reconnecting()): return None if (host._pending_host_addition and on_add_reconnection is None): @@ -2503,13 +2501,15 @@ def _start_reconnector(self, host, is_host_addition, # on_remove() will clear and cancel this handler; if removal wins, do # not leave retry work attached to a terminal Host object. with host.lock: - if host._is_removed: + if not self._lifecycle.down_is_current(host, down_event_generation): return None - if is_host_addition: - host._pending_host_addition = True + if restart and (host.is_up is not False or + host._currently_handling_node_up or + host.is_currently_reconnecting()): + return None + if is_host_addition or host._pending_host_addition: + self._lifecycle.remember_addition(host, on_add_reconnection) is_host_addition = host._pending_host_addition - if is_host_addition and on_add_reconnection is not None: - host._pending_host_addition_callback = on_add_reconnection reconnector = _HostReconnectionHandler( host, conn_factory, is_host_addition, on_add, self.on_up, self.scheduler, schedule, @@ -2526,6 +2526,7 @@ def _start_reconnector(self, host, is_host_addition, return reconnector @run_in_executor + @_lifecycle.event def on_down_potentially_blocking(self, host, is_host_addition, down_event_generation, on_add_reconnection=None): @@ -2538,94 +2539,67 @@ def on_down_potentially_blocking(self, host, is_host_addition, # on_down() reports successful dispatch to callers that rely on a # reconnection being started. try: - if not self._dispatch_down_callback( - host, down_event_generation, - self.profile_manager.on_down): + if not self._lifecycle.dispatch_current( + host, self.profile_manager.on_down, + generation=down_event_generation): return except Exception: log.exception("Error in load balancing policy down handler for host %s", host) try: - if not self._dispatch_down_callback( - host, down_event_generation, - self.control_connection.on_down): + if not self._lifecycle.dispatch_current( + host, self.control_connection.on_down, + generation=down_event_generation): return except Exception: log.exception("Error in control connection down handler for host %s", host) for session in tuple(self.sessions): try: - if not self._dispatch_down_callback( - host, down_event_generation, session.on_down): + if not self._lifecycle.dispatch_current( + host, session.on_down, generation=down_event_generation): return except Exception: log.exception("Error marking host %s down in session", host) for listener in self.listeners: try: - if not self._dispatch_down_callback( - host, down_event_generation, listener.on_down): + if not self._lifecycle.dispatch_current( + host, listener.on_down, generation=down_event_generation): return except Exception: log.exception("Error in host state listener down handler for host %s", host) - # Pool cleanup and listener callbacks above may race a newer DOWN - # generation or host removal. Pair the final check with handler - # installation so stale work cannot publish a reconnector. with host.lock: - if down_event_generation != host._down_event_generation: + if not self._lifecycle.down_is_current(host, down_event_generation): return - if on_add_reconnection is None: - self._start_reconnector(host, is_host_addition) - else: - self._start_reconnector( - host, is_host_addition, - on_add_reconnection=on_add_reconnection) + # Policy callbacks and schedule construction run unlocked. The + # final installation rechecks this generation under host.lock. + self._start_reconnector( + host, is_host_addition, + on_add_reconnection=on_add_reconnection, + down_event_generation=down_event_generation) finally: - with host.lock: - if down_event_generation == host._down_event_generation: - host._currently_handling_node_down = False - - @staticmethod - def _dispatch_down_callback(host, down_event_generation, callback): - """Invoke a current DOWN callback before removal can overtake it.""" - # Removal waits for this callback before notifying on_remove. User - # callbacks must not wait on cluster executor work; #382 tracks moving - # lifecycle ordering to queued work instead of blocking a worker. - with host._down_callbacks_lock: - with host.lock: - if (host._is_removed or - down_event_generation != host._down_event_generation): - return False - callback(host) - return True + self._lifecycle.finish_down(host, down_event_generation) @run_in_executor def _restart_reconnector(self, host, is_host_addition, down_event_generation, on_add_reconnection=None): try: - # Match the old synchronous path's ordering with on_up(): if the - # host recovered or is still being brought up while this task was - # queued, there is nothing to restart. A failed UP transition - # starts its own reconnector before releasing the UP guard, which - # must not be replaced by this stale task. with host.lock: - if (down_event_generation == host._down_event_generation and - host.is_up is False and - not host._currently_handling_node_up and - not host.is_currently_reconnecting()): - if on_add_reconnection is None: - self._start_reconnector(host, is_host_addition) - else: - self._start_reconnector( - host, is_host_addition, - on_add_reconnection=on_add_reconnection) + if (not self._lifecycle.down_is_current(host, down_event_generation) or + host.is_up is not False or + host._currently_handling_node_up or + host.is_currently_reconnecting()): + return + self._start_reconnector( + host, is_host_addition, + on_add_reconnection=on_add_reconnection, + down_event_generation=down_event_generation, restart=True) finally: - with host.lock: - if down_event_generation == host._down_event_generation: - host._currently_handling_node_down = False + self._lifecycle.finish_down(host, down_event_generation) def on_down(self, host, is_host_addition, expect_host_to_be_down=False, - on_add_reconnection=None): + on_add_reconnection=None, _pool_failure=None): """ Intended for internal use only. @@ -2634,78 +2608,44 @@ def on_down(self, host, is_host_addition, expect_host_to_be_down=False, if self.is_shutdown or self.allow_control_connection_query_fallback == ControlConnectionQueryFallback.SkipPoolCreation: return False - restart_reconnector = False - with host.lock: - if host._is_removed: - return False - - was_up = host.is_up - - # ignore down signals if we have open pools to the host - # this is to avoid closing pools when a control connection host became isolated - if self._discount_down_events and self.profile_manager.distance(host) != HostDistance.IGNORED: - connected = False - for session in tuple(self.sessions): - pool_states = session.get_pool_state() - pool_state = pool_states.get(host) - if pool_state: - connected |= pool_state['open_count'] > 0 - if connected: - return False - - host.set_down() - if (not was_up and - (host.is_currently_reconnecting() or - host._currently_handling_node_down)): - return False - - if not was_up and not expect_host_to_be_down: - # A terminal failure may have released this down host's old - # handler. Start a new retry cycle without repeating the DOWN - # notifications that were sent on the original transition. - restart_reconnector = True - - host._currently_handling_node_down = True - host._down_event_generation += 1 - down_event_generation = host._down_event_generation - - if restart_reconnector: - if on_add_reconnection is None: - future = self._restart_reconnector( - host, is_host_addition, down_event_generation) - else: - future = self._restart_reconnector( - host, is_host_addition, down_event_generation, - on_add_reconnection=on_add_reconnection) - if future is None: - with host.lock: - if down_event_generation == host._down_event_generation: - host._currently_handling_node_down = False - # Restarting the host reconnector does not dispatch the DOWN - # callbacks that reconnect the control connection. - return False - - log.warning("Host %s has been marked down", host) + # Custom distance policies may refresh topology. Do not invoke them + # under the Host state lock needed by terminal publication. + connected = False + if (self._discount_down_events and + self.profile_manager.distance(host) != HostDistance.IGNORED): + for session in tuple(self.sessions): + pool_state = session.get_pool_state().get(host) + if pool_state: + connected |= pool_state['open_count'] > 0 - if on_add_reconnection is None: - future = self.on_down_potentially_blocking( - host, is_host_addition, down_event_generation) - else: - future = self.on_down_potentially_blocking( - host, is_host_addition, down_event_generation, - on_add_reconnection=on_add_reconnection) - if future is None: - with host.lock: - if down_event_generation == host._down_event_generation: - host._currently_handling_node_down = False - return future is not None + return self._lifecycle.dispatch_down( + host, is_host_addition, connected, expect_host_to_be_down, + on_add_reconnection, _pool_failure) def on_add(self, host, refresh_nodes=True, reconcile_pools_on_failure=False, set_up_if_ignored=False): + def deliver(): + try: + self._on_add(host, refresh_nodes, reconcile_pools_on_failure, + set_up_if_ignored) + except Exception: + if self._is_current_host(host): + with host.lock: + if not host._is_removed: + host._endpoint_add_retry_pending = True + self._schedule_failed_endpoint_add_retry(host) + raise + self._lifecycle.submit_current(host, deliver) + + def _on_add(self, host, refresh_nodes=True, + reconcile_pools_on_failure=False, + set_up_if_ignored=False): if self.is_shutdown: return + if not self._is_current_host(host): + return on_add_reconnection = None if reconcile_pools_on_failure: # The initial replacement add runs inside a topology refresh, but @@ -2716,22 +2656,26 @@ def on_add(self, host, refresh_nodes=True, reconcile_pools_on_failure=True, set_up_if_ignored=True) - with host.lock: - if host._is_removed: - return - host._pending_host_addition = True - if on_add_reconnection is not None: - host._pending_host_addition_callback = on_add_reconnection + if not self._lifecycle.begin_addition(host, on_add_reconnection): + return log.debug("Handling new host %r and notifying listeners", host) self.profile_manager.on_add(host) + if not self._is_current_host(host): + return self.control_connection.on_add(host, refresh_nodes) + if not self._is_current_host(host): + return distance = self.profile_manager.distance(host) + if not self._is_current_host(host): + return if distance != HostDistance.IGNORED: self._prepare_all_queries(host) log.debug("Done preparing queries for new host %r", host) + if not self._is_current_host(host): + return if distance == HostDistance.IGNORED: log.debug("Not adding connection pool for new host %r because the " @@ -2767,6 +2711,13 @@ def future_completed(future): self._finalize_add(host) return + if not reconcile_pools_on_failure: + # A healthy sibling pool can discount DOWN. Once aggregate + # ADD has failed it no longer owns ordinary pool creation; + # let that session reconcile without tearing down live pools. + self._lifecycle.release_addition(host) + return + # A same-endpoint replacement removes its predecessor without # reconciling pools because doing so can race this addition. If # the replacement fails, _finalize_add() cannot provide that @@ -2795,11 +2746,8 @@ def future_completed(future): # as terminal. Supersede that task as well as any handler # it managed to install, and allow a later DOWN event to # start fresh handling. - with host.lock: - host._down_event_generation += 1 - host._currently_handling_node_down = False - old_reconnector = \ - host.get_and_set_reconnection_handler(None) + old_reconnector = self._lifecycle.supersede_down(host) + self._lifecycle.release_addition(host) if old_reconnector: old_reconnector.cancel() else: @@ -2862,42 +2810,44 @@ def future_completed(future): # when the future is already complete, so registering inside the loop # can finalize the host while later sessions are still unscheduled. for future in tuple(futures): - future.add_done_callback(future_completed) + future.add_done_callback( + lambda future: self._lifecycle.submit(future_completed, future)) + @_lifecycle.event def _finalize_add(self, host, set_up=True): - with host.lock: - if host._is_removed: - return - if set_up: - host.set_up() - host._pending_host_addition = False - host._pending_host_addition_callback = None + if not self._lifecycle.mark_ready(host, set_up): + return for listener in self.listeners: - listener.on_add(host) + try: + if not self._lifecycle.dispatch_current(host, listener.on_add): + return + except Exception: + log.exception("Error in host addition listener for %s", 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() def on_remove(self, host, trigger_reconciliation=True, - suppress_callback_errors=False): - """Retire a host, optionally isolating replacement callback failures.""" - if self.is_shutdown: - return - - self._retire_host(host) - self._notify_host_removed(host, trigger_reconciliation, - suppress_callback_errors) + suppress_callback_errors=False, refresh_nodes=True): + """Retire a host and queue its ordered removal notifications.""" + self._lifecycle.retire( + host, trigger_reconciliation=trigger_reconciliation, + refresh_nodes=refresh_nodes) def _notify_host_removed(self, host, trigger_reconciliation, - suppress_callback_errors): + suppress_callback_errors, refresh_nodes=True): if self.is_shutdown: return log.debug("[cluster] Removing host %s", host) def notify(callback, *args, **kwargs): + if self.is_shutdown: + return False if not suppress_callback_errors: callback(*args, **kwargs) return True @@ -2915,40 +2865,37 @@ def notify(callback, *args, **kwargs): notify(session.remove_pool, host) for listener in self.listeners: notify(listener.on_remove, host) - notify(self.control_connection.on_remove, host, - trigger_reconciliation=trigger_reconciliation) + options = dict(trigger_reconciliation=trigger_reconciliation) + if not refresh_nodes: + options["refresh_nodes"] = False + notify(self.control_connection.on_remove, host, **options) reconnection_handler = host.get_and_set_reconnection_handler(None) 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 + try: + reconnection_handler.cancel() + except Exception: + log.exception("Error cancelling removed host reconnector") def signal_connection_failure(self, host, connection_exc, is_host_addition, expect_host_to_be_down=False, - on_add_reconnection=None): + on_add_reconnection=None, _pool_failure=None): """Return whether the failure caused DOWN handling to be dispatched.""" + if _pool_failure is not None and not _pool_failure.is_current(): + return False is_down = host.signal_connection_failure(connection_exc) if not is_down: return False + options = {} if _pool_failure is None else {'_pool_failure': _pool_failure} if on_add_reconnection is None: return self.on_down( - host, is_host_addition, expect_host_to_be_down) + host, is_host_addition, expect_host_to_be_down, **options) return self.on_down( host, is_host_addition, expect_host_to_be_down, - on_add_reconnection=on_add_reconnection) + on_add_reconnection=on_add_reconnection, **options) + @_lifecycle.transaction def add_host(self, endpoint, datacenter=None, rack=None, signal=True, refresh_nodes=True, host_id=None, reconcile_pools_on_failure=False, @@ -2974,34 +2921,9 @@ def add_host(self, endpoint, datacenter=None, rack=None, signal=True, candidate = Host(endpoint, self.conviction_policy_factory, datacenter, rack, host_id=host_id) - host, new, replaced_host = self.metadata._add_or_replace_host( - candidate, allow_replace=signal) - if not new and host.endpoint != candidate.endpoint: - if not signal: - raise ValueError("Endpoint replacement requires host addition signaling") - if not replace_existing_id: - raise ValueError("Endpoint replacement is disabled for this host addition") - return self.replace_host_endpoint( - host, candidate.endpoint, datacenter, rack), False - - if replaced_host is not None: - log.info("Cassandra host %s removed", replaced_host) - # The replacement is already visible. Its addition owns pool - # reconciliation, so removal must not start a nested refresh. - try: - self.on_remove(replaced_host, trigger_reconciliation=False, - suppress_callback_errors=True) - except Exception: - log.exception("Error notifying removal of replaced host %s", - replaced_host) - reconcile_pools_on_failure = True - if new and signal: - log.info("New Cassandra host %r discovered", host) - self.on_add( - host, refresh_nodes, - reconcile_pools_on_failure=reconcile_pools_on_failure) - - return host, new + return self._lifecycle.publish_host( + candidate, signal, refresh_nodes, reconcile_pools_on_failure, + replace_existing_id) def replace_host_endpoint(self, host, endpoint, datacenter=None, rack=None): """Move a host ID to a new endpoint with a fresh Host lifecycle. @@ -3013,6 +2935,7 @@ def replace_host_endpoint(self, host, endpoint, datacenter=None, rack=None): return self.replace_hosts_endpoints( [(host, endpoint, datacenter, rack)])[0] + @_lifecycle.transaction def replace_hosts_endpoints(self, moves): """Publish all moving Hosts before retiring and adding their routes.""" replacements = [] @@ -3044,35 +2967,14 @@ def replace_hosts_endpoints(self, moves): connection._control_connection_host = host bound_host = host break - try: - self.metadata.replace_host_endpoints(replacements) - except ValueError as exc: - raise _EndpointMoveRejected(str(exc)) from exc reconnect_control = connection is not None and any( bound_host is host or (bound_host is None and control_connection._connection_matches_host(connection, host)) for host, _ in replacements) - for host, replacement in replacements: - log.info("Cassandra host %s moved from %s to %s", host.host_id, - host.endpoint, replacement.endpoint) - self.on_remove(host, trigger_reconciliation=False, - suppress_callback_errors=True) - try: - for _, replacement in replacements: - try: - self.on_add(replacement, refresh_nodes=False, - reconcile_pools_on_failure=True) - except Exception: - log.exception("Error notifying addition of moved host %s", - replacement) - with replacement.lock: - replacement._endpoint_add_retry_pending = True - self._schedule_failed_endpoint_add_retry(replacement) - finally: - if reconnect_control: - control_connection.reconnect() + self._lifecycle.publish_replacements( + replacements, control_connection.reconnect if reconnect_control else None) return [replacement for _, replacement in replacements] def _schedule_failed_endpoint_add_retry(self, host): @@ -3091,53 +2993,35 @@ def _schedule_failed_endpoint_add_retry(self, host): log.exception("Error scheduling addition retry for moved host %s", host) + @_lifecycle.event def retry_failed_endpoint_add(self, host): """Retry a published endpoint replacement whose synchronous add failed.""" - # Removal uses this same lock to make the old lifecycle terminal. - # Keep policy/pool setup ordered before that removal or skip it. - with host._down_callbacks_lock: - if self.metadata.get_host_by_host_id(host.host_id) is not host: + # The queued retry may have been superseded by retirement. + if self.metadata.get_host_by_host_id(host.host_id) is not host: + return + with host.lock: + if (host._is_removed or + not getattr(host, '_endpoint_add_retry_pending', False)): return + # Claim this retry so overlapping topology refreshes do not + # run policy and pool setup for the same Host at once. + host._endpoint_add_retry_pending = False + try: + self.on_add(host, refresh_nodes=False, + reconcile_pools_on_failure=True) + except Exception: + log.exception("Error retrying addition of moved host %s", host) with host.lock: - if (host._is_removed or - not getattr(host, '_endpoint_add_retry_pending', False)): - return - # Claim this retry so overlapping topology refreshes do not - # run policy and pool setup for the same Host at once. - host._endpoint_add_retry_pending = False - try: - self.on_add(host, refresh_nodes=False, - reconcile_pools_on_failure=True) - except Exception: - log.exception("Error retrying addition of moved host %s", host) - with host.lock: - if not host._is_removed: - host._endpoint_add_retry_pending = True - self._schedule_failed_endpoint_add_retry(host) + if not host._is_removed: + host._endpoint_add_retry_pending = True + self._schedule_failed_endpoint_add_retry(host) def remove_host(self, host, trigger_reconciliation=True, - suppress_callback_errors=False): - """ - Called when the control connection observes that a node has left the - ring. Intended for internal use only. - - ``trigger_reconciliation`` may be disabled when an immediately - following host addition owns pool and control-connection follow-up. - ``suppress_callback_errors`` keeps replacement cleanup moving if a - removal listener raises. - """ - 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: - 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) + suppress_callback_errors=False, refresh_nodes=True): + """Unpublish a current host and queue its lifecycle cleanup.""" + self._lifecycle.remove( + host, trigger_reconciliation=trigger_reconciliation, + refresh_nodes=refresh_nodes) def register_listener(self, listener): """ @@ -3280,7 +3164,13 @@ def refresh_user_aggregate_metadata(self, keyspace, aggregate, max_schema_agreem def refresh_nodes(self, force_token_rebuild=False): """ - Synchronously refresh the node list and token metadata + Synchronously publish refreshed node-list and token metadata. + + DC/rack location transitions and lifecycle notifications are delivered + in order outside metadata locks. + If another notification is running (including this call from a + listener), this method can return before the queued notifications run. + A newer accepted refresh supersedes an older in-flight snapshot. `force_token_rebuild` can be used to rebuild the token map metadata, even if no new nodes are discovered. @@ -4559,8 +4449,14 @@ def _handle_pool_creation_failure(self, host, generation, host, generation, expected_pool, endpoint)): return self._stale_pool_creation_result_locked( host, expected_pool, intent, distance) - self._pool_generations[host] = object() - dispatch_failure() + failure_generation = object() + self._pool_generations[host] = failure_generation + reservation = _lifecycle.PoolFailureReservation( + self, host, failure_generation, expected_pool, intent, endpoint) + # Conviction/distance callbacks can refresh topology. Retain ownership + # in a token and recheck it atomically with DOWN admission instead of + # holding session/host locks through arbitrary policy code. + dispatch_failure(reservation) return owning_result def add_or_renew_pool(self, host, is_host_addition, @@ -4730,7 +4626,7 @@ def run_add_or_renew_pool(): connection_errors = [] - def dispatch_connection_failure(): + def dispatch_connection_failure(reservation): connection_exc = connection_errors[0] log.warning( "Failed to create connection pool for new host %s:", @@ -4740,12 +4636,13 @@ def dispatch_connection_failure(): if on_add_reconnection is None: self.cluster.signal_connection_failure( host, connection_exc, is_host_addition, - expect_host_to_be_down=True) + expect_host_to_be_down=True, _pool_failure=reservation) else: self.cluster.signal_connection_failure( host, connection_exc, is_host_addition, expect_host_to_be_down=True, - on_add_reconnection=on_add_reconnection) + on_add_reconnection=on_add_reconnection, + _pool_failure=reservation) try: new_pool = HostConnection(host, distance, self) @@ -4753,14 +4650,16 @@ def dispatch_connection_failure(): authentication_exc = ConnectionException( str(auth_exc), endpoint=host) - def dispatch_authentication_failure(): + def dispatch_authentication_failure(reservation): if on_add_reconnection is None: self.cluster.signal_connection_failure( - host, authentication_exc, is_host_addition) + host, authentication_exc, is_host_addition, + _pool_failure=reservation) else: self.cluster.signal_connection_failure( host, authentication_exc, is_host_addition, - on_add_reconnection=on_add_reconnection) + on_add_reconnection=on_add_reconnection, + _pool_failure=reservation) # Replacement aggregation uses None to distinguish a # non-retryable authentication failure from other falsey @@ -4826,18 +4725,20 @@ def callback(pool, errors): new_pool._set_keyspace_for_all_conns(keyspace, callback) set_keyspace_event.wait(self.cluster.connect_timeout) if not set_keyspace_event.is_set() or errors_returned: - def dispatch_keyspace_failure(): + def dispatch_keyspace_failure(reservation): log.warning( "Failed setting keyspace for pool after " "keyspace changed during connect: %s", errors_returned) if on_add_reconnection is None: - self.cluster.on_down(host, is_host_addition) + self.cluster.on_down( + host, is_host_addition, _pool_failure=reservation) else: self.cluster.on_down( host, is_host_addition, expect_host_to_be_down=True, - on_add_reconnection=on_add_reconnection) + on_add_reconnection=on_add_reconnection, + _pool_failure=reservation) try: return self._handle_pool_creation_failure( @@ -4949,7 +4850,7 @@ def update_created_pools(self, excluded_host=None, hosts=None): if self.is_shutdown: break with host.lock: - if host._is_removed: + if host._is_removed or host._lifecycle_pending: continue # In-flight on_add/on_up owns pool creation for a down host; # superseding it would settle its result False with no retry. @@ -5574,6 +5475,7 @@ def __init__(self, cluster, timeout, self._token_meta_enabled = token_meta_enabled self._schema_meta_page_size = schema_meta_page_size + self._lifecycle = self._cluster._lifecycle self._lock = RLock() self._schema_agreement_lock = Lock() @@ -5904,6 +5806,7 @@ def _try_connect(self, endpoint): connection, self._timeout) + revision = self._lifecycle.reserve_refresh() sel_peers = self._get_peers_query(self.PeersQueryType.PEERS, connection) sel_local = self._SELECT_LOCAL if self._token_meta_enabled else self._SELECT_LOCAL_NO_TOKENS peers_query = QueryMessage(query=maybe_add_timeout_to_query(sel_peers, self._metadata_request_timeout), @@ -5926,7 +5829,8 @@ def _try_connect(self, endpoint): peers_query, timeout=self._timeout) shared_results = (peers_result, local_result) - self._refresh_node_list_and_token_map(connection, preloaded_results=shared_results) + self._refresh_node_list_and_token_map( + connection, preloaded_results=shared_results, revision=revision) self._refresh_schema(connection, preloaded_results=shared_results, schema_agreement_wait=-1) except Exception: connection.close() @@ -6185,7 +6089,9 @@ def refresh_node_list_and_token_map(self, force_token_rebuild=False): return False def _refresh_node_list_and_token_map(self, connection, preloaded_results=None, - force_token_rebuild=False): + force_token_rebuild=False, revision=None): + if revision is None: + revision = self._lifecycle.reserve_refresh(force_token_rebuild) if preloaded_results: log.debug("[control connection] Refreshing node list and token map using preloaded results") peers_result = preloaded_results[0] @@ -6206,36 +6112,60 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None, peers_result, local_result = connection.wait_for_responses( peers_query, local_query, timeout=self._timeout) - peers_result = dict_factory(peers_result.column_names, peers_result.parsed_rows) - - partitioner = None + rows = dict_factory(peers_result.column_names, peers_result.parsed_rows) + local_row = None + if local_result.parsed_rows: + local_row = dict_factory(local_result.column_names, + local_result.parsed_rows)[0] + rows.insert(0, local_row) + prepared = [(row, self._cluster.endpoint_factory.create(row)) + for row in rows if self._is_valid_peer(row)] + with self._lifecycle.transaction(): + if not self._lifecycle.accept_refresh(revision): + if not self._lifecycle.closed: + self._bind_stale_candidate(connection, prepared, local_row) + return + force_token_rebuild |= self._lifecycle.force_rebuild_required(revision) + self._apply_node_list_and_token_map( + connection, prepared, local_row, force_token_rebuild) + if force_token_rebuild: + self._lifecycle.complete_token_rebuild(revision) + + def _bind_stale_candidate(self, connection, prepared, local_row): + """Bind socket-local identity without publishing an older topology.""" + if vars(connection).get('_control_connection_host') is not None: + return + for row, endpoint in prepared: + if row is not local_row: + continue + host_id = row.get('host_id') + host = self._cluster.metadata.get_host_by_host_id(host_id) + original = getattr(connection, 'original_endpoint', connection.endpoint) + if (host is not None and not host._is_removed and + host.endpoint in (endpoint, original)): + connection._control_connection_host_id = host_id + connection._control_connection_host = host + return + # A candidate socket whose observed identity/route was superseded + # must be retried, not installed against the new endpoint owner. + raise ConnectionException("Control connection topology snapshot is stale", + connection.endpoint) + + def _apply_node_list_and_token_map( + self, connection, prepared, local_row, force_token_rebuild): + partitioner = local_row.get('partitioner') if local_row else None token_map = {} - found_host_ids = set() found_endpoints = set() - - local_row = None - if local_result.parsed_rows: - local_rows = dict_factory(local_result.column_names, local_result.parsed_rows) - local_row = local_rows[0] - cluster_name = local_row["cluster_name"] - self._cluster.metadata.cluster_name = cluster_name - - partitioner = local_row.get("partitioner") - tokens = local_row.get("tokens", None) - - peers_result.insert(0, local_row) + if local_row: + self._cluster.metadata.cluster_name = local_row['cluster_name'] # Check metadata.partitioner to see if we haven't built anything yet. If # every node in the cluster was in the contact points, we won't discover # any new nodes, so we need this additional check. (See PYTHON-90) should_rebuild_token_map = force_token_rebuild or self._cluster.metadata.partitioner is None accepted_rows = [] - for row in peers_result: - if not self._is_valid_peer(row): - continue - - factory_endpoint = self._cluster.endpoint_factory.create(row) + for row, factory_endpoint in prepared: host_id = row.get("host_id") # Use the factory endpoint for duplicate detection even when a Unix @@ -6303,7 +6233,7 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None, if moves: try: replacements = self._cluster.replace_hosts_endpoints(moves) - except _EndpointMoveRejected: + except _lifecycle.EndpointMoveRejected: log.warning( "[control connection] Endpoint moves changed during refresh; skipping stale batch", exc_info=True) @@ -6455,7 +6385,7 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None, getattr(host, '_endpoint_add_retry_pending', False)): # Query preparation in on_add can block on a slow node. # Keep the topology refresh moving while the scheduler runs - # the retry under the host's callback lock. + # the retry through the lifecycle notification queue. self._cluster.scheduler.schedule_unique( 0, self._cluster.retry_failed_endpoint_add, host) @@ -6463,10 +6393,8 @@ 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) - # 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) + self._cluster.remove_host( + old_host, refresh_nodes=False, suppress_callback_errors=True) log.debug("[control connection] Finished fetching ring info") if partitioner and should_rebuild_token_map: @@ -6517,16 +6445,7 @@ def _is_valid_peer(row): return True def _update_location_info(self, host, datacenter, rack): - if host.datacenter == datacenter and host.rack == rack: - return False - - # If the dc/rack information changes, we need to update the load balancing policy. - # For that, we remove and re-add the node against the policy. Not the most elegant, and assumes - # that the policy will update correctly, but in practice this should work. - self._cluster.profile_manager.on_down(host) - host.set_location_info(datacenter, rack) - self._cluster.profile_manager.on_up(host) - return True + return self._lifecycle.update_location(host, datacenter, rack) def _delay_for_event_type(self, event_type, delay_window): # this serves to order processing correlated events (received within the window) @@ -6863,7 +6782,7 @@ def on_add(self, host, refresh_nodes=True): if refresh_nodes: self.refresh_node_list_and_token_map(force_token_rebuild=True) - def on_remove(self, host, trigger_reconciliation=True): + def on_remove(self, host, trigger_reconciliation=True, refresh_nodes=True): if not trigger_reconciliation: # Same-endpoint replacement reconciliation is already running on # either the current connection or a candidate that its caller @@ -6876,7 +6795,7 @@ def on_remove(self, host, trigger_reconciliation=True): log.debug("[control connection] Control connection host (%s) is being removed. Reconnecting", host) # refresh will be done on reconnect self.reconnect() - else: + elif refresh_nodes: self.refresh_node_list_and_token_map(force_token_rebuild=True) def get_connections(self): diff --git a/cassandra/metadata.py b/cassandra/metadata.py index d1c4ef8f9a..300a4e393b 100644 --- a/cassandra/metadata.py +++ b/cassandra/metadata.py @@ -16,7 +16,6 @@ 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 @@ -363,31 +362,16 @@ 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. """ - 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 + 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 if replaced_host is not None: self._tablets.drop_tablets_by_host_id(replaced_host.host_id) @@ -411,41 +395,48 @@ def replace_host_endpoints(self, replacements): replacements = list(replacements) source_ids = set() destinations = set() - # 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 + 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 _update_host_location(self, host, datacenter, rack): + """Publish a current Host's location with fresh replica-map caches.""" + with self._hosts_lock: + if self._hosts.get(host.host_id) is not host: + return False + host.set_location_info(datacenter, rack) + token_map = self.token_map + if token_map is not None: + # A concurrent lazy replica calculation can finish on the old + # TokenMap without repopulating caches on the current one. + self.token_map = TokenMap( + token_map.token_class, token_map.token_to_host_owner, + token_map.ring, self) + return True def remove_host(self, host): with self._hosts_lock: diff --git a/cassandra/policies.py b/cassandra/policies.py index 61db69d6f6..1c33deb31e 100644 --- a/cassandra/policies.py +++ b/cassandra/policies.py @@ -81,6 +81,19 @@ class HostDistance(object): class HostStateListener(object): + """Receive ordered host lifecycle notifications. + + Notifications run without topology or host callback locks. Metadata may + already reflect a newer topology while a callback is running. An active + callback can finish after its Host is retired; stale queued UP, DOWN and + ADD work is discarded, and removal is delivered before replacement ADD. + + A listener may call :meth:`~cassandra.cluster.Cluster.refresh_nodes` to + publish metadata synchronously. Notifications caused by that refresh are + queued until the current notification finishes, rather than delivered + recursively. Return promptly: waiting for another notification from the + same cluster prevents its FIFO dispatcher from making progress. + """ def on_up(self, host): """ Called when a node is marked up. """ @@ -90,9 +103,9 @@ def on_down(self, host): """ Called when a node is marked down. - Return promptly. Host removal waits for an in-progress DOWN callback, - so do not wait synchronously for work submitted to the cluster executor. - This applies to registered listeners and custom load-balancing policies. + Return promptly; later lifecycle notifications wait for this callback. + Metadata removal itself does not wait. This also applies to custom + load-balancing policies. """ raise NotImplementedError() diff --git a/cassandra/pool.py b/cassandra/pool.py index a1e2f49d78..ce55c1882c 100644 --- a/cassandra/pool.py +++ b/cassandra/pool.py @@ -160,9 +160,9 @@ class Host(object): _currently_handling_node_up = False _currently_handling_node_down = False _down_event_generation = 0 + _lifecycle_pending = False _pending_host_addition = False _pending_host_addition_callback = None - _down_callbacks_lock = None sharding_info = None @@ -183,10 +183,6 @@ def __init__(self, endpoint, conviction_policy_factory, datacenter=None, rack=No self.conviction_policy = conviction_policy_factory(self) self.set_location_info(datacenter, rack) self.lock = RLock() - # DOWN callbacks call into policies and sessions. Keep them ordered - # with terminal removal without holding the Host state lock across - # those external callbacks. - self._down_callbacks_lock = RLock() @property def address(self): diff --git a/docs/api/cassandra/cluster.rst b/docs/api/cassandra/cluster.rst index f782677f1c..96fc7c47c0 100644 --- a/docs/api/cassandra/cluster.rst +++ b/docs/api/cassandra/cluster.rst @@ -232,3 +232,43 @@ Clusters and Sessions :members: .. autoexception:: UserTypeDoesNotExist () + +Topology publication and lifecycle notifications +------------------------------------------------ + +:meth:`Cluster.refresh_nodes` synchronously publishes node-list and token +metadata. DC/rack attribute changes are delivered with their policy transition +(remove from the old location, update the Host, add at the new location), and +can lag node-list publication while a notification is running. The method does +not wait for an already-running lifecycle notification. +Policies, session pool management and registered host listeners receive queued +notifications in order, with no topology or host callback lock held while they +run. A replacement's removal notifications precede its addition notifications. + +A listener can call ``refresh_nodes()`` synchronously, including while another +thread removes its host. The nested refresh publishes metadata and appends its +notifications; it does not recursively invoke a replacement's addition inside +an unfinished removal callback. Consequently metadata can be newer than the +notifications currently being delivered, and pool availability can lag metadata +publication. A newly published host remains reserved for its addition operation +until lifecycle initialization finishes. + +Retirement is terminal for a Host object. An already-running callback may +finish after retirement, but queued stale UP, DOWN and ADD work is discarded. +A callback must return before later notifications can run; do not wait for +another lifecycle notification from the same cluster. Callback exceptions are +logged and do not strand the notification queue. + +Topology queries can overlap. A snapshot older than a newer accepted refresh +is discarded instead of overwriting the newer topology. This orders client +refresh requests; it does not establish a server-side topology version. + +Shutdown first closes lifecycle admission and discards queued notifications. +This coordinator step does not wait for an active listener. An external +``shutdown()`` call still waits for executor workers during executor teardown; +a listener calling it from an executor worker skips that wait to avoid joining +itself. Already-admitted user +callbacks may finish, but later notifications, refresh publication and late +pool-completion callbacks cannot restart the closed lifecycle. Pending addition +and location reservations are cleared, and current or retired host reconnectors +are cancelled. diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index de6cef302c..36d4286711 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -514,8 +514,7 @@ def remove_then_fail(*_args, **_kwargs): cluster.sessions = (session,) with self.assertLogs('cassandra.cluster', level='ERROR'): - with self.assertRaisesRegex(RuntimeError, 'pool scheduling failed'): - cluster.on_up(host) + cluster.on_up(host) assert host.is_up is False assert not host._currently_handling_node_up @@ -523,7 +522,7 @@ def remove_then_fail(*_args, **_kwargs): cluster.control_connection.on_down.assert_not_called() cluster._start_reconnector.assert_not_called() - def test_remove_waits_for_inflight_up_completion(self): + def test_remove_orders_notifications_after_inflight_up_completion(self): cluster = Cluster() self.addCleanup(cluster.shutdown) cluster.profile_manager = Mock() @@ -541,7 +540,6 @@ def test_remove_waits_for_inflight_up_completion(self): cluster.sessions = (session,) callback_started = Event() release_callback = Event() - removal_attempted = Event() removal_finished = Event() callback_order = [] @@ -556,21 +554,6 @@ def on_listener_up(_host): 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(): @@ -584,8 +567,9 @@ def remove_host(): try: assert callback_started.wait(5) remove_thread.start() - assert removal_attempted.wait(5) - assert not removal_finished.is_set() + assert removal_finished.wait(5) + assert host._is_removed + assert callback_order == [] finally: release_callback.set() completion_thread.join(5) @@ -595,10 +579,10 @@ def remove_host(): 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() + session.update_created_pools.assert_not_called() assert host._is_removed - def test_remove_releases_callback_lock_before_notifications(self): + def test_remove_releases_coordination_lock_before_notifications(self): cluster = Cluster() self.addCleanup(cluster.shutdown) cluster.control_connection.on_remove = Mock() @@ -606,20 +590,22 @@ def test_remove_releases_callback_lock_before_notifications(self): host_id=uuid.uuid4()) cluster.metadata.add_or_return_host(host) lock_acquired = Event() + observations = [] def on_listener_remove(_host): - assert cluster.metadata.get_host_by_host_id(host.host_id) is None - assert host._is_removed + observations.extend(( + cluster.metadata.get_host_by_host_id(host.host_id) is None, + host._is_removed)) def acquire_callback_lock(): - with host._down_callbacks_lock: + with cluster._lifecycle.transaction(): lock_acquired.set() lock_thread = Thread(target=acquire_callback_lock) lock_thread.start() - assert lock_acquired.wait(2) + observations.append(lock_acquired.wait(2)) lock_thread.join(2) - assert not lock_thread.is_alive() + observations.append(not lock_thread.is_alive()) listener = Mock() listener.on_remove.side_effect = on_listener_remove @@ -627,11 +613,12 @@ def acquire_callback_lock(): cluster.remove_host(host) + assert observations == [True, True, True, True] 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): + def test_replacement_orders_notifications_after_inflight_up_completion(self): cluster = Cluster() self.addCleanup(cluster.shutdown) cluster.profile_manager = Mock() @@ -650,7 +637,6 @@ def test_replacement_waits_for_inflight_up_completion(self): cluster.sessions = (session,) callback_started = Event() release_callback = Event() - replacement_attempted = Event() replacement_finished = Event() callback_order = [] @@ -665,21 +651,6 @@ def on_listener_up(_host): 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(): @@ -693,9 +664,9 @@ def replace_host(): 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() + assert replacement_finished.wait(5) + assert host._is_removed + assert callback_order == [] finally: release_callback.set() completion_thread.join(5) @@ -709,7 +680,7 @@ 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): + def test_endpoint_move_orders_notifications_after_inflight_up_completion(self): cluster = Cluster() self.addCleanup(cluster.shutdown) cluster.profile_manager = Mock() @@ -728,7 +699,7 @@ def test_endpoint_move_waits_for_inflight_up_completion(self): cluster.sessions = (session,) callback_started = Event() release_callback = Event() - move_attempted = Event() + move_finished = Event() callback_order = [] def on_listener_up(_host): @@ -742,21 +713,6 @@ def on_listener_up(_host): 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 = [] @@ -766,14 +722,17 @@ def move_host(): host, DefaultEndPoint('127.0.0.2')) except Exception as exc: move_errors.append(exc) + finally: + move_finished.set() 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 + assert move_finished.wait(5) + assert host._is_removed + assert callback_order == [] finally: release_callback.set() completion_thread.join(5) @@ -1961,7 +1920,9 @@ def test_fresh_down_supersedes_queued_reconnector_restart(self): cluster.control_connection.on_down.assert_called_once_with(self.host) session.on_down.assert_called_once_with(self.host) listener.on_down.assert_called_once_with(self.host) - cluster._start_reconnector.assert_called_once_with(self.host, False) + cluster._start_reconnector.assert_called_once_with( + self.host, False, on_add_reconnection=None, + down_event_generation=self.host._down_event_generation) assert not self.host._currently_handling_node_down def test_queued_reconnector_restart_skips_pending_up(self): @@ -2134,7 +2095,7 @@ def on_add(host): listener.on_add.side_effect = on_add - with pytest.raises(RuntimeError, match='listener failed'): + with self.assertLogs('cassandra.cluster', level='ERROR'): cluster.on_add(self.host) assert not self.host._pending_host_addition @@ -2165,7 +2126,9 @@ def test_repeated_down_waits_for_pending_down_processing(self): down_task, *args = cluster.executor.submit.call_args.args down_task(*args) - cluster._start_reconnector.assert_called_once_with(self.host, False) + cluster._start_reconnector.assert_called_once_with( + self.host, False, on_add_reconnection=None, + down_event_generation=self.host._down_event_generation) assert not self.host._currently_handling_node_down def test_failed_down_submission_releases_pending_state(self): @@ -5269,7 +5232,8 @@ def test_down_handling_continues_when_the_control_connection_raises(self): self.cluster.profile_manager.on_down.assert_called_once_with(self.host) self.cluster._start_reconnector.assert_called_once_with( - self.host, False) + self.host, False, on_add_reconnection=None, + down_event_generation=self.host._down_event_generation) def test_down_handling_continues_when_a_session_raises(self): del self.cluster.on_down_potentially_blocking # use the real method @@ -5286,7 +5250,8 @@ def test_down_handling_continues_when_a_session_raises(self): following_session.on_down.assert_called_once_with(self.host) self.cluster._start_reconnector.assert_called_once_with( - self.host, False) + self.host, False, on_add_reconnection=None, + down_event_generation=self.host._down_event_generation) def test_down_handling_continues_when_a_listener_raises(self): del self.cluster.on_down_potentially_blocking # use the real method @@ -5303,7 +5268,8 @@ def test_down_handling_continues_when_a_listener_raises(self): following_listener.on_down.assert_called_once_with(self.host) self.cluster._start_reconnector.assert_called_once_with( - self.host, False) + self.host, False, on_add_reconnection=None, + down_event_generation=self.host._down_event_generation) def test_reentrant_removal_stops_remaining_down_callbacks(self): del self.cluster.on_down_potentially_blocking # use the real method @@ -5351,14 +5317,13 @@ def test_reentrant_policy_removal_stops_later_policy_callbacks(self): self.cluster.control_connection.on_down.assert_not_called() self.cluster._start_reconnector.assert_not_called() - def test_remove_waits_for_inflight_down_callback(self): + def test_remove_orders_notifications_after_inflight_down_callback(self): del self.cluster.on_down_potentially_blocking # use the real method self.cluster.metadata.add_or_return_host(self.host) self.cluster.profile_manager.on_down = Mock() self.cluster._start_reconnector = Mock() callback_started = Event() release_callback = Event() - removal_attempted = Event() removal_finished = Event() callback_order = [] @@ -5376,21 +5341,6 @@ def on_control_down(_host): self.host.set_down() self.host._currently_handling_node_down = True - callback_lock = self.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() - - self.host._down_callbacks_lock = AttemptTrackingLock() body = Cluster.on_down_potentially_blocking.__wrapped__ down_thread = Thread( target=body, @@ -5407,13 +5357,13 @@ def remove_host(): down_thread.start() assert callback_started.wait(5) remove_thread.start() - assert removal_attempted.wait(5) - removal_was_blocked = not removal_finished.is_set() + assert removal_finished.wait(5) + assert self.host._is_removed + assert callback_order == [] release_callback.set() down_thread.join(5) remove_thread.join(5) - assert removal_was_blocked assert not down_thread.is_alive() assert not remove_thread.is_alive() assert callback_order.index('down') < callback_order.index('remove') diff --git a/tests/unit/test_control_connection.py b/tests/unit/test_control_connection.py index 4c76465c2b..6209d153d1 100644 --- a/tests/unit/test_control_connection.py +++ b/tests/unit/test_control_connection.py @@ -24,17 +24,18 @@ from cassandra import (AuthenticationFailed, OperationTimedOut, SchemaTargetType, SchemaChangeType, UnresolvableContactPoints) +from cassandra._lifecycle import LifecycleCoordinator, EndpointMoveRejected from cassandra.protocol import ResultMessage, RESULT_KIND_ROWS from cassandra.cluster import (Cluster, ControlConnection, _Scheduler, ProfileManager, EXEC_PROFILE_DEFAULT, ExecutionProfile, ControlConnectionQueryFallback, - NoHostAvailable, _ControlReconnectionHandler, - _EndpointMoveRejected) + NoHostAvailable, _ControlReconnectionHandler) from cassandra.pool import Host, _ReconnectionHandler from cassandra.metadata import Metadata from cassandra.connection import (ConnectionException, EndPoint, DefaultEndPoint, - DefaultEndPointFactory, UnixSocketEndPoint) + DefaultEndPointFactory, UnixSocketEndPoint, + SniEndPointFactory, ClientRoutesEndPointFactory) from cassandra.policies import (DCAwareRoundRobinPolicy, HostDistance, SimpleConvictionPolicy, RoundRobinPolicy, ConstantReconnectionPolicy, @@ -124,6 +125,12 @@ def replace_host_endpoints(self, replacements): self.hosts[replacement.host_id] = replacement self._host_id_by_endpoint[replacement.endpoint] = replacement.host_id + def _update_host_location(self, host, datacenter, rack): + if self.hosts.get(host.host_id) is not host: + return False + host.set_location_info(datacenter, rack) + return True + def remove_host(self, host): if self.hosts.get(host.host_id) is not host: return False @@ -157,6 +164,7 @@ class MockCluster(object): is_shutdown = False def __init__(self): + self._lifecycle = LifecycleCoordinator(self) self.metadata = MockMetadata() self.sessions = [] self.added_hosts = [] @@ -176,8 +184,10 @@ def add_host(self, endpoint, datacenter, rack, signal=False, return host, True def remove_host(self, host, trigger_reconciliation=True, - suppress_callback_errors=False): - pass + suppress_callback_errors=False, refresh_nodes=True): + if self.metadata.remove_host(host): + host._is_removed = True + self.metadata.removed_hosts.append(host) def replace_host_endpoint(self, host, endpoint, datacenter, rack): replacement = Host( @@ -1925,7 +1935,9 @@ def add_after_move(*args, **kwargs): self.control_connection._refresh_node_list_and_token_map(self.connection) - self.cluster.remove_host.assert_not_called() + self.cluster.remove_host.assert_called_once_with( + self.cluster.metadata.get_host_by_host_id(HOST_ID_3), + refresh_nodes=False, suppress_callback_errors=True) replacement = self.cluster.metadata.get_host_by_host_id(HOST_ID_2) assert replacement is not moving_host assert self.cluster.metadata.get_host(moved_endpoint) is replacement @@ -1942,7 +1954,7 @@ def test_rejected_move_batch_continues_new_hosts_and_token_rebuild(self): self.connection.wait_for_responses.return_value = _node_meta_results( self.connection.local_results, self.connection.peer_results) self.cluster.replace_hosts_endpoints = Mock( - side_effect=_EndpointMoveRejected('stale source')) + side_effect=EndpointMoveRejected('stale source')) self.cluster.remove_host = Mock( side_effect=lambda host, **kwargs: self.cluster.metadata.remove_host(host)) @@ -1974,7 +1986,7 @@ def test_rejected_batch_keeps_tokens_for_concurrently_replaced_host(self): def reject_after_concurrent_move(_moves): self.cluster.metadata.replace_host_endpoint(old, current) - raise _EndpointMoveRejected('stale source') + raise EndpointMoveRejected('stale source') self.cluster.replace_hosts_endpoints = Mock( side_effect=reject_after_concurrent_move) @@ -2819,54 +2831,119 @@ def test_refresh_nodes_and_tokens_add_host(self): assert self.cluster.added_hosts[0].rack == "rack1" assert self.cluster.added_hosts[0].host_id == HOST_ID_4 + def test_topology_removal_runs_full_lifecycle_for_endpoint_factories(self): + for factory in (DefaultEndPointFactory(), + SniEndPointFactory('proxy', 9042), + ClientRoutesEndPointFactory(Mock(), 9042)): + with self.subTest(factory=type(factory).__name__): + cluster = Cluster(load_balancing_policy=RoundRobinPolicy()) + self.addCleanup(cluster.shutdown) + cluster.control_connection.shutdown() + cluster.endpoint_factory = factory.configure(cluster) + cluster._prepare_all_queries = Mock() + connection = MockConnection() + rows = [dict(zip(connection.local_results[0], + connection.local_results[1][0]))] + rows += [dict(zip(connection.peer_results[0], row)) + for row in connection.peer_results[1]] + hosts = [] + for row in rows: + host, _ = cluster.add_host( + factory.create(row), signal=False, + host_id=row['host_id']) + host.set_up() + cluster.profile_manager.on_add(host) + hosts.append(host) + old = hosts[1] + pool = Mock(host=old, is_shutdown=False) + session = new_session_with_pool_state({old: pool}) + session.cluster = cluster + session.is_shutdown = False + + def submit(fn, *args, **kwargs): + fn(*args, **kwargs) + future = Future() + future.set_result(True) + return future + + session.submit = submit + session.add_or_renew_pool = Mock( + return_value=submit(lambda: None)) + session.update_created_pools = Mock() + session.shutdown = Mock() + cluster.sessions.add(session) + listener = Mock() + cluster.register_listener(listener) + connection.endpoint = hosts[0].endpoint + connection.original_endpoint = hosts[0].endpoint + connection.peer_results[1][0][-1] = HOST_ID_4 + connection.wait_for_responses = Mock( + return_value=_node_meta_results( + connection.local_results, connection.peer_results)) + control = ControlConnection(cluster, 1, 2, 0, 0) + cluster.control_connection = control + control._connection = connection + + assert control.refresh_node_list_and_token_map() + assert control.refresh_node_list_and_token_map() + + replacement = cluster.metadata.get_host_by_host_id(HOST_ID_4) + assert old._is_removed and not old.is_up + assert old not in session._pools + pool.shutdown.assert_called_once_with() + listener.on_remove.assert_called_once_with(old) + listener.on_add.assert_called_once_with(replacement) + assert cluster.metadata.get_host_by_host_id(old.host_id) is None + if old.endpoint != replacement.endpoint: + assert cluster.metadata.get_host(old.endpoint) is None + assert cluster.metadata.get_host(replacement.endpoint) is replacement + assert old not in list( + cluster.profile_manager.default.load_balancing_policy.make_query_plan()) + assert session.update_created_pools.called + assert connection.wait_for_responses.call_count == 2 + + def test_topology_removal_finishes_after_listener_failure(self): + cluster = Cluster(load_balancing_policy=RoundRobinPolicy()) + self.addCleanup(cluster.shutdown) + cluster.control_connection.shutdown() + for host_id, address in ((HOST_ID_1, '192.168.1.0'), + (HOST_ID_2, '192.168.1.1'), + (HOST_ID_3, '192.168.1.2')): + host, _ = cluster.add_host( + DefaultEndPoint(address), signal=False, host_id=host_id) + host.set_up() + cluster.profile_manager.on_add(host) + old = cluster.metadata.get_host_by_host_id(HOST_ID_3) + reconnector = Mock() + old.get_and_set_reconnection_handler(reconnector) + listener = Mock() + listener.on_remove.side_effect = RuntimeError('listener failed') + cluster.register_listener(listener) + control = ControlConnection(cluster, 1, 2, 0, 0) + cluster.control_connection = control + control._connection = self.connection + control.on_remove = Mock() + del self.connection.peer_results[1][1] + cluster.metadata.rebuild_token_map = Mock() + + assert control.refresh_node_list_and_token_map() + + assert old._is_removed + assert cluster.metadata.get_host_by_host_id(HOST_ID_3) is None + listener.on_remove.assert_called_once_with(old) + control.on_remove.assert_called_once_with( + old, trigger_reconciliation=True, refresh_nodes=False) + reconnector.cancel.assert_called_once_with() + assert not old.is_currently_reconnecting() + cluster.metadata.rebuild_token_map.assert_called_once() + assert old not in cluster.metadata.rebuild_token_map.call_args.args[1] + def test_refresh_nodes_and_tokens_remove_host(self): del self.connection.peer_results[1][1] self.control_connection.refresh_node_list_and_token_map() 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): diff --git a/tests/unit/test_lifecycle_coordination.py b/tests/unit/test_lifecycle_coordination.py new file mode 100644 index 0000000000..6ce833a544 --- /dev/null +++ b/tests/unit/test_lifecycle_coordination.py @@ -0,0 +1,926 @@ +# Copyright DataStax, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from concurrent.futures import Future +from threading import Event, Lock, Thread +from unittest.mock import Mock + +import pytest + +from cassandra.cluster import Cluster, ControlConnection +from cassandra._lifecycle import LifecycleCoordinator +from cassandra.connection import DefaultEndPoint +from tests.unit.test_control_connection import ( + HOST_ID_1, HOST_ID_2, HOST_ID_3, MockConnection, _node_meta_results) + + +@pytest.fixture +def topology(): + cluster = Cluster() + cluster.control_connection.shutdown() + cluster._prepare_all_queries = Mock() + cluster.profile_manager = Mock() + cluster.profile_manager.distance.return_value = 0 + cluster._start_reconnector = Mock() + for host_id, address in ((HOST_ID_1, '192.168.1.0'), + (HOST_ID_2, '192.168.1.1'), + (HOST_ID_3, '192.168.1.2')): + host, _ = cluster.add_host(DefaultEndPoint(address), signal=False, + host_id=host_id) + host.set_up() + connection = MockConnection() + control = ControlConnection(cluster, 1, 2, 0, 0) + control._connection = connection + cluster.control_connection = control + yield cluster, connection + cluster.shutdown() + + +@pytest.mark.parametrize('event', ['up', 'down', 'add']) +def test_listener_refresh_during_concurrent_removal(topology, event): + cluster, connection = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + entered = Event() + published = Event() + refreshed = Event() + order = [] + errors = [] + listener = Mock() + + def callback(current): + assert current is host + entered.set() + assert published.wait(5) + assert current._is_removed + cluster.refresh_nodes() + refreshed.set() + order.append(event) + + getattr(listener, 'on_' + event).side_effect = callback + listener.on_remove.side_effect = lambda host: order.append('remove') + cluster.register_listener(listener) + + def deliver(): + try: + if event == 'up': + result = Future() + result.set_result(True) + cluster._on_up_future_completed(host, {result}, [], Lock(), result) + elif event == 'down': + Cluster.on_down_potentially_blocking.__wrapped__( + cluster, host, False, host._down_event_generation) + else: + host._endpoint_add_retry_pending = True + cluster.retry_failed_endpoint_add(host) + except Exception as exc: + errors.append(exc) + + worker = Thread(target=deliver) + worker.start() + try: + assert entered.wait(5) + del connection.peer_results[1][1] + cluster.refresh_nodes() + assert cluster.metadata.get_host_by_host_id(HOST_ID_3) is None + assert host._is_removed + assert order == [] + finally: + published.set() + worker.join(5) + assert not worker.is_alive() + assert not errors + assert refreshed.is_set() + assert order == [event, 'remove'] + cluster._start_reconnector.assert_not_called() + + +def test_reentrant_removal_publishes_before_delivering_replacement(topology): + cluster, connection = topology + old = cluster.metadata.get_host_by_host_id(HOST_ID_3) + old_row = connection.peer_results[1].pop() + order = [] + replacements = [] + + def remove(host): + assert host is old + order.append('remove-start') + connection.peer_results[1].append(old_row) + cluster.refresh_nodes() + replacement = cluster.metadata.get_host_by_host_id(HOST_ID_3) + assert replacement is not old + assert replacement._lifecycle_pending + replacements.append(replacement) + assert order == ['remove-start'] + order.append('remove-end') + + def add(host): + assert host is replacements[0] + order.append('add') + + cluster.profile_manager.on_remove.side_effect = remove + cluster.profile_manager.on_add.side_effect = add + cluster.refresh_nodes() + assert order == ['remove-start', 'remove-end', 'add'] + assert old._is_removed + assert not replacements[0]._lifecycle_pending + assert replacements[0].is_up + + +def test_older_query_cannot_overwrite_newer_topology(topology): + cluster, connection = topology + queried = Event() + release = Event() + old_connection = MockConnection() + del old_connection.peer_results[1][1] + old_result = _node_meta_results(old_connection.local_results, + old_connection.peer_results) + errors = [] + + def query(*args, **kwargs): + queried.set() + assert release.wait(5) + return old_result + + old_connection.wait_for_responses.side_effect = query + + def refresh_old(): + try: + cluster.control_connection._refresh_node_list_and_token_map(old_connection) + except Exception as exc: + errors.append(exc) + + worker = Thread(target=refresh_old) + worker.start() + try: + assert queried.wait(5) + current = cluster.metadata.get_host_by_host_id(HOST_ID_3) + cluster.refresh_nodes() + finally: + release.set() + worker.join(5) + assert not worker.is_alive() + assert not errors + assert cluster.metadata.get_host_by_host_id(HOST_ID_3) is current + assert not current._is_removed + cluster.profile_manager.on_remove.assert_not_called() + + +def test_dispatcher_releases_ownership_after_callback_errors(): + coordinator = LifecycleCoordinator() + observed = [] + + def fail(): + coordinator.submit(observed.append, 'nested') + raise RuntimeError('listener failed') + + with coordinator.transaction(): + coordinator.submit(fail) + coordinator.submit(observed.append, 'next') + coordinator.submit(observed.append, 'last') + assert observed == ['next', 'nested', 'last'] + + +def test_publication_exception_still_delivers_committed_notifications(): + coordinator = LifecycleCoordinator() + observed = [] + with pytest.raises(ValueError): + with coordinator.transaction(): + coordinator.submit(observed.append, 'committed') + raise ValueError('later row rejected') + assert observed == ['committed'] + + +def test_notification_does_not_hold_coordination_mutex(): + coordinator = LifecycleCoordinator() + acquired = Event() + observations = [] + + def acquire(): + with coordinator.transaction(): + acquired.set() + + def callback(): + thread = Thread(target=acquire) + thread.start() + observations.append(acquired.wait(5)) + thread.join(5) + observations.append(not thread.is_alive()) + + coordinator.submit(callback) + assert acquired.is_set() + assert observations == [True, True] + + +@pytest.mark.parametrize('with_future', [False, True]) +def test_successful_up_releases_pending_addition_reservation(topology, with_future): + cluster, connection = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + host.set_down() + host._lifecycle_pending = True + host._pending_host_addition = True + if with_future: + completed = Future() + completed.set_result(True) + cluster._on_up_future_completed(host, {completed}, [], Lock(), completed) + else: + cluster.on_up(host) + assert host.is_up + assert not host._lifecycle_pending + assert not host._pending_host_addition + + +@pytest.mark.parametrize('policy_kind', ['dc', 'rack']) +def test_location_transition_removes_old_policy_membership(topology, policy_kind): + from cassandra.policies import DCAwareRoundRobinPolicy, RackAwareRoundRobinPolicy + cluster, connection = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + host.set_location_info('dc1', 'rack1') + policy = (DCAwareRoundRobinPolicy('dc1', used_hosts_per_remote_dc=1) + if policy_kind == 'dc' else + RackAwareRoundRobinPolicy('dc1', 'rack1', used_hosts_per_remote_dc=1)) + policy.populate(cluster, [host]) + cluster.profile_manager.on_down.side_effect = policy.on_down + cluster.profile_manager.on_up.side_effect = policy.on_up + cluster.profile_manager.on_remove.side_effect = policy.on_remove + with cluster._lifecycle.transaction(): + assert cluster.control_connection._update_location_info(host, 'dc2', 'rack2') + assert host.datacenter == 'dc1' + assert host.datacenter == 'dc2' + assert host.rack == 'rack2' + assert list(policy.make_query_plan()).count(host) == 1 + cluster.remove_host(host, refresh_nodes=False) + assert host not in list(policy.make_query_plan()) + + +@pytest.mark.parametrize('custom_callback', ['distance', 'reset']) +def test_custom_policy_refresh_does_not_hold_host_state_lock(topology, custom_callback): + cluster, connection = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + entered = Event() + published = Event() + returned = Event() + errors = [] + + def callback(*args): + entered.set() + assert published.wait(5) + cluster.refresh_nodes() + returned.set() + return 0 + + if custom_callback == 'distance': + cluster.profile_manager.distance.side_effect = callback + del cluster._start_reconnector + operation = lambda: Cluster.on_down_potentially_blocking.__wrapped__( + cluster, host, False, host._down_event_generation) + else: + host.conviction_policy.reset = callback + operation = lambda: cluster._finalize_add(host) + + def run(): + try: + operation() + except Exception as exc: + errors.append(exc) + + worker = Thread(target=run) + worker.start() + try: + assert entered.wait(5) + del connection.peer_results[1][1] + cluster.refresh_nodes() + assert host._is_removed + finally: + published.set() + worker.join(5) + assert not worker.is_alive() + assert not errors + assert returned.is_set() + assert host._is_removed and not host.is_up + assert host._reconnection_handler is None + + +def test_newer_location_reversal_supersedes_queued_change(topology): + cluster, connection = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + host.set_location_info('dc1', 'rack1') + control = cluster.control_connection + with cluster._lifecycle.transaction(): + assert control._update_location_info(host, 'dc2', 'rack2') + assert control._update_location_info(host, 'dc1', 'rack1') + assert (host.datacenter, host.rack) == ('dc1', 'rack1') + assert host._pending_location is None + + +def test_location_delivery_replaces_stale_replica_cache(topology): + from cassandra.metadata import KeyspaceMetadata, Murmur3Token + cluster, connection = topology + metadata = cluster.metadata + host = metadata.get_host_by_host_id(HOST_ID_3) + host.set_location_info('dc1', 'rack1') + metadata.keyspaces['ks'] = KeyspaceMetadata( + 'ks', True, 'NetworkTopologyStrategy', {'dc2': 1}) + with cluster._lifecycle.transaction(): + cluster.control_connection._update_location_info(host, 'dc2', 'rack2') + metadata.rebuild_token_map('Murmur3Partitioner', {host: ['0']}) + old_map = metadata.token_map + assert old_map.get_replicas('ks', Murmur3Token(0)) == [] + assert metadata.token_map is not old_map + assert metadata.token_map.get_replicas('ks', Murmur3Token(0)) == [host] + # Even late publication of old calculated caches cannot poison the new map. + old_map.tokens_to_hosts_by_ks['ks'] = {Murmur3Token(0): []} + assert metadata.token_map.get_replicas('ks', Murmur3Token(0)) == [host] + + +def test_repeated_location_destination_keeps_latest_reservation(topology): + cluster, connection = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + host.set_location_info('dc1', 'rack1') + control = cluster.control_connection + requested = [] + + def on_up(current): + if current.datacenter == 'dc3' and not requested: + requested.append(True) + assert control._update_location_info(host, 'dc3', 'rack1') + + cluster.profile_manager.on_up.side_effect = on_up + with cluster._lifecycle.transaction(): + for dc in ('dc2', 'dc3', 'dc2'): + assert control._update_location_info(host, dc, 'rack1') + assert host.datacenter == 'dc3' + assert host._pending_location is None + + +@pytest.mark.parametrize('deferred', [False, True]) +def test_notification_submission_never_returns_callback_result(deferred): + coordinator = LifecycleCoordinator() + observed = [] + + def callback(): + observed.append('delivered') + return 'not a synchronous result' + + if deferred: + with coordinator.transaction(): + assert coordinator.submit(callback) is None + assert observed == [] + else: + assert coordinator.submit(callback) is None + assert observed == ['delivered'] + + +def test_queued_addition_checks_identity_at_delivery(topology): + cluster, _ = topology + old = cluster.metadata.get_host_by_host_id(HOST_ID_3) + callback = Mock() + with cluster._lifecycle.transaction(): + cluster._lifecycle.submit_current(old, callback) + replacement = cluster.replace_host_endpoint( + old, DefaultEndPoint('192.168.1.9')) + assert replacement.host_id == old.host_id + assert old._is_removed + callback.assert_not_called() + assert cluster.metadata.get_host_by_host_id(old.host_id) is replacement + + +def test_retirement_during_readiness_reset_cannot_mark_host_ready(topology): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + cluster._lifecycle.reserve_addition(host) + host.conviction_policy.reset = Mock( + side_effect=lambda: cluster.remove_host(host)) + assert not cluster._lifecycle.mark_ready(host) + assert host._is_removed + assert not host.is_up + assert not host._lifecycle_pending + + +@pytest.mark.parametrize('event', ['location', 'up', 'down', 'add']) +def test_shutdown_discards_pending_notifications(topology, event): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + original_location = (host.datacenter, host.rack) + host.set_down() + with cluster._lifecycle.transaction(): + if event == 'location': + cluster._lifecycle.update_location(host, 'later_dc', 'later_rack') + elif event == 'up': + cluster.on_up(host) + elif event == 'down': + Cluster.on_down_potentially_blocking.__wrapped__( + cluster, host, False, host._down_event_generation) + else: + cluster.on_add(host) + cluster.shutdown() + cluster.profile_manager.on_up.assert_not_called() + cluster.profile_manager.on_down.assert_not_called() + cluster.profile_manager.on_add.assert_not_called() + assert (host.datacenter, host.rack) == original_location + assert host._pending_location is None + assert not host._lifecycle_pending + assert not cluster._lifecycle._events + callback = Mock() + cluster._lifecycle.submit(callback) + callback.assert_not_called() + + +def test_shutdown_from_executor_listener_does_not_join_itself(topology): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + listener = Mock() + shutdown_returned = Event() + + def shutdown_from_listener(_host): + cluster.shutdown() + shutdown_returned.set() + + listener.on_up.side_effect = shutdown_from_listener + cluster.register_listener(listener) + result = Future() + result.set_result(True) + finished = cluster.executor.submit( + cluster._on_up_future_completed, host, {result}, [], Lock(), result) + finished.result(timeout=5) + assert cluster.is_shutdown + assert shutdown_returned.is_set() + assert cluster._lifecycle.closed + assert not host._currently_handling_node_up + listener.on_up.assert_called_once_with(host) + + +def test_location_listener_shutdown_stops_location_publication(topology): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + original_location = (host.datacenter, host.rack) + cluster.profile_manager.on_down.side_effect = lambda _: cluster.shutdown() + cluster._lifecycle.update_location(host, 'later_dc', 'later_rack') + assert (host.datacenter, host.rack) == original_location + cluster.profile_manager.on_up.assert_not_called() + assert host._pending_location is None + + +def test_late_up_pool_completion_after_shutdown_is_discarded(topology): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + host.set_down() + session = Mock() + pool_ready = Future() + session.add_or_renew_pool.return_value = pool_ready + cluster.sessions.add(session) + listener = Mock() + cluster.register_listener(listener) + cluster.on_up(host) + assert host._currently_handling_node_up + cluster.shutdown() + pool_ready.set_result(True) + listener.on_up.assert_not_called() + assert not host.is_up + assert not host._currently_handling_node_up + session.update_created_pools.assert_not_called() + + +def test_repeated_retirement_delivers_removal_once(topology): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + listener = Mock() + cluster.register_listener(listener) + with cluster._lifecycle.transaction(): + cluster.remove_host(host, refresh_nodes=False) + cluster.on_remove(host, refresh_nodes=False) + cluster.remove_host(host, refresh_nodes=False) + listener.on_remove.assert_called_once_with(host) + assert host._is_removed + assert not host._lifecycle_pending + + +def test_shutdown_cancels_retired_reconnector_before_queued_removal(topology): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + handler = Mock() + host._reconnection_handler = handler + with cluster._lifecycle.transaction(): + cluster.remove_host(host, refresh_nodes=False) + cluster.shutdown() + handler.cancel.assert_called_once_with() + assert host._reconnection_handler is None + + +def test_close_does_not_wait_for_active_notification(): + coordinator = LifecycleCoordinator() + entered = Event() + release = Event() + later = Mock() + + def callback(): + entered.set() + assert release.wait(5) + coordinator.submit(later) + + worker = Thread(target=coordinator.submit, args=(callback,)) + worker.start() + try: + assert entered.wait(5) + coordinator.submit(later) + coordinator.close() + assert worker.is_alive() + assert coordinator.closed + finally: + release.set() + worker.join(5) + assert not worker.is_alive() + later.assert_not_called() + + +def test_refresh_result_after_shutdown_cannot_publish(topology): + cluster, connection = topology + queried = Event() + release = Event() + old_query = connection.wait_for_responses.side_effect + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + del connection.peer_results[1][1] + result = _node_meta_results(connection.local_results, connection.peer_results) + + def query(*args, **kwargs): + queried.set() + assert release.wait(5) + return result + + connection.wait_for_responses.side_effect = query + apply_topology = Mock( + wraps=cluster.control_connection._apply_node_list_and_token_map) + cluster.control_connection._apply_node_list_and_token_map = apply_topology + errors = [] + + def refresh(): + try: + cluster.control_connection._refresh_node_list_and_token_map(connection) + except Exception as exc: + errors.append(exc) + + worker = Thread(target=refresh) + worker.start() + try: + assert queried.wait(5) + cluster.shutdown() + finally: + release.set() + worker.join(5) + connection.wait_for_responses.side_effect = old_query + assert not worker.is_alive() + assert not errors + apply_topology.assert_not_called() + assert cluster.metadata.get_host_by_host_id(host.host_id) is host + cluster.profile_manager.on_remove.assert_not_called() + + +@pytest.mark.parametrize('event', ['up', 'add']) +def test_policy_shutdown_prevents_remaining_lifecycle_work(topology, event): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + host.set_down() + cluster.profile_manager.distance.side_effect = lambda _: (cluster.shutdown() or 0) + cluster.control_connection.on_add = Mock() + getattr(cluster, 'on_' + event)(host) + cluster._prepare_all_queries.assert_not_called() + cluster.profile_manager.on_up.assert_not_called() + assert not host._currently_handling_node_up + + +def test_failed_ordinary_add_releases_reservation_without_dropping_healthy_pool(topology): + from tests.unit.utils import new_session_with_pool_state + + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + host.is_up = None + cluster.control_connection.on_add = Mock() + healthy_pool = Mock(host=host, is_shutdown=False) + good = new_session_with_pool_state({host: healthy_pool}) + bad = new_session_with_pool_state() + for session, succeeded in ((good, True), (bad, False)): + session.cluster = cluster + session._profile_manager = cluster.profile_manager + session.is_shutdown = False + session.shutdown = Mock() + result = Future() + result.set_result(succeeded) + session.add_or_renew_pool = Mock(return_value=result) + session.get_pool_state = Mock( + return_value={host: {'open_count': int(succeeded)}}) + cluster.sessions.add(session) + cluster.on_add(host) + assert not host._lifecycle_pending + assert not cluster.on_down(host, True, expect_host_to_be_down=True) + bad.add_or_renew_pool.reset_mock() + bad.update_created_pools(hosts=(host,)) + assert bad.add_or_renew_pool.call_count == 1 + assert good._pools[host] is healthy_pool + healthy_pool.shutdown.assert_not_called() + + +def test_newer_refresh_preserves_pending_forced_token_rebuild(topology): + cluster, connection = topology + cluster.refresh_nodes(force_token_rebuild=True) + old_ring = list(cluster.metadata.token_map.ring) + queried = Event() + release = Event() + older = MockConnection() + old_result = _node_meta_results(older.local_results, older.peer_results) + + def query(*args, **kwargs): + queried.set() + assert release.wait(5) + return old_result + + older.wait_for_responses.side_effect = query + errors = [] + + def force_refresh(): + try: + cluster.control_connection._refresh_node_list_and_token_map( + older, force_token_rebuild=True) + except Exception as exc: + errors.append(exc) + + worker = Thread(target=force_refresh) + worker.start() + try: + assert queried.wait(5) + connection.peer_results[1][1][5] = ['300', '400', '500'] + cluster.refresh_nodes() + new_ring = list(cluster.metadata.token_map.ring) + assert new_ring != old_ring + finally: + release.set() + worker.join(5) + assert not worker.is_alive() + assert not errors + assert list(cluster.metadata.token_map.ring) == new_ring + + +def test_stale_unix_candidate_binds_without_republishing_topology(topology): + from cassandra.connection import UnixSocketEndPoint + + cluster, _ = topology + candidate = MockConnection() + candidate.original_endpoint = candidate.endpoint = UnixSocketEndPoint( + '/tmp/lifecycle-candidate.sock') + revision = cluster._lifecycle.reserve_refresh() + cluster.refresh_nodes() + current = cluster.metadata.get_host_by_host_id(HOST_ID_1) + token_map = cluster.metadata.token_map + cluster.control_connection._refresh_node_list_and_token_map( + candidate, preloaded_results=_node_meta_results( + candidate.local_results, candidate.peer_results), revision=revision) + assert candidate._control_connection_host is current + assert cluster.control_connection._connection_matches_host(candidate, current) + assert cluster.metadata.token_map is token_map + + +def test_stale_candidate_cannot_bind_to_new_route(topology): + from cassandra.connection import ConnectionException + + cluster, _ = topology + candidate = MockConnection() + revision = cluster._lifecycle.reserve_refresh() + cluster.refresh_nodes() + old = cluster.metadata.get_host_by_host_id(HOST_ID_1) + replacement = cluster.replace_host_endpoint( + old, DefaultEndPoint('192.168.1.100')) + with pytest.raises(ConnectionException, match='snapshot is stale'): + cluster.control_connection._refresh_node_list_and_token_map( + candidate, preloaded_results=_node_meta_results( + candidate.local_results, candidate.peer_results), revision=revision) + assert vars(candidate).get('_control_connection_host') is None + assert cluster.metadata.get_host_by_host_id(HOST_ID_1) is replacement + + +def test_duplicate_up_does_not_schedule_another_pool_creation(topology): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + host.set_down() + cluster.control_connection.on_up = Mock() + pool_ready = Future() + session = Mock() + session.add_or_renew_pool.return_value = pool_ready + cluster.sessions.add(session) + cluster.on_up(host) + cluster.on_up(host) + session.add_or_renew_pool.assert_called_once_with(host, is_host_addition=False) + assert host._currently_handling_node_up + pool_ready.set_result(True) + assert host.is_up + assert not host._currently_handling_node_up + + +def test_superseded_down_does_not_reach_later_callbacks(topology): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + generation = host._down_event_generation + cluster.profile_manager.on_down.side_effect = ( + lambda current: cluster._lifecycle.supersede_down(current)) + cluster.control_connection.on_down = Mock() + session = Mock() + cluster.sessions.add(session) + listener = Mock() + cluster.register_listener(listener) + Cluster.on_down_potentially_blocking.__wrapped__( + cluster, host, False, generation) + cluster.profile_manager.on_down.assert_called_once_with(host) + cluster.control_connection.on_down.assert_not_called() + session.on_down.assert_not_called() + listener.on_down.assert_not_called() + cluster._start_reconnector.assert_not_called() + + +def test_later_failed_force_does_not_hide_earlier_force_intent(topology): + cluster, connection = topology + cluster.refresh_nodes(force_token_rebuild=True) + old_ring = list(cluster.metadata.token_map.ring) + older_force = cluster._lifecycle.reserve_refresh(force_token_rebuild=True) + middle_refresh = cluster._lifecycle.reserve_refresh() + failing_connection = MockConnection() + failing_connection.wait_for_responses.side_effect = RuntimeError('query failed') + with pytest.raises(RuntimeError, match='query failed'): + cluster.control_connection._refresh_node_list_and_token_map( + failing_connection, force_token_rebuild=True) + connection.peer_results[1][1][5] = ['300', '400', '500'] + cluster.control_connection._refresh_node_list_and_token_map( + connection, preloaded_results=_node_meta_results( + connection.local_results, connection.peer_results), revision=middle_refresh) + middle_ring = list(cluster.metadata.token_map.ring) + assert middle_ring != old_ring + cluster.control_connection._refresh_node_list_and_token_map( + connection, preloaded_results=_node_meta_results( + connection.local_results, connection.peer_results), revision=older_force) + assert list(cluster.metadata.token_map.ring) == middle_ring + connection.peer_results[1][1][5] = ['600', '700', '800'] + cluster.refresh_nodes() + assert list(cluster.metadata.token_map.ring) != middle_ring + + +def test_pool_failure_policy_refresh_runs_without_session_or_host_lock(topology): + from tests.unit.utils import new_session_with_pool_state + + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + session = new_session_with_pool_state() + session.cluster = cluster + session.shutdown = Mock() + session.get_pool_state = Mock(return_value={}) + cluster.sessions.add(session) + cluster.on_down_potentially_blocking = Mock(return_value=Future()) + with session._lock: + intent = session._claim_pool_intent_locked(host) + generation = object() + session._pool_generations[host] = generation + observations = [] + + def distance(_host): + def publication(): + with cluster._lifecycle.transaction(): + session_locked = session._lock.acquire(timeout=2) + host_locked = host.lock.acquire(timeout=2) + observations.extend((session_locked, host_locked)) + if host_locked: + host.lock.release() + if session_locked: + session._lock.release() + + worker = Thread(target=publication) + worker.start() + worker.join(5) + observations.append(not worker.is_alive()) + cluster.refresh_nodes() + return 0 + + cluster.profile_manager.distance.side_effect = distance + session._handle_pool_creation_failure( + host, generation, None, intent, 0, + lambda reservation: cluster.on_down( + host, False, _pool_failure=reservation), False) + assert observations == [True, True, True] + cluster.on_down_potentially_blocking.assert_called_once() + + +def test_pool_failure_rechecks_ownership_after_policy_callback(topology): + from tests.unit.utils import new_session_with_pool_state + + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + session = new_session_with_pool_state() + session.cluster = cluster + session.shutdown = Mock() + session.get_pool_state = Mock(return_value={}) + cluster.sessions.add(session) + cluster.on_down_potentially_blocking = Mock(return_value=Future()) + with session._lock: + intent = session._claim_pool_intent_locked(host) + generation = object() + session._pool_generations[host] = generation + winner = Mock(host=host, is_shutdown=False) + + def publish_new_pool(_host): + with session._lock: + with host.lock: + session._pools[host] = winner + session._pool_generations[host] = object() + return 0 + + cluster.profile_manager.distance.side_effect = publish_new_pool + session._handle_pool_creation_failure( + host, generation, None, intent, 0, + lambda reservation: cluster.on_down( + host, False, _pool_failure=reservation), False) + assert host.is_up + assert session._pools[host] is winner + cluster.on_down_potentially_blocking.assert_not_called() + winner.shutdown.assert_not_called() + + +def test_shutdown_flag_racing_replacement_does_not_skip_old_retirement(topology): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + handler = Mock() + host._reconnection_handler = handler + with cluster._lifecycle.transaction(): + cluster.metadata.remove_host(host) + cluster.is_shutdown = True + cluster.on_remove(host) + assert host._is_removed + cluster._shutdown(wait_for_executor=True) + handler.cancel.assert_called_once_with() + + +def test_removal_cancels_detached_reconnector_even_if_shutdown_wins(topology): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + handler = Mock() + host._reconnection_handler = handler + take_handler = host.get_and_set_reconnection_handler + + def take_then_shutdown(replacement): + previous = take_handler(replacement) + if previous is handler: + cluster.shutdown() + return previous + + host.get_and_set_reconnection_handler = take_then_shutdown + cluster.remove_host(host, refresh_nodes=False) + handler.cancel.assert_called_once_with() + + +def test_down_is_queued_before_executor_handoff(topology): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + cluster._discount_down_events = False + cluster.control_connection.on_down = Mock() + cluster.control_connection.on_up = Mock() + original_submit = cluster.executor.submit + scheduled = [] + + def delayed_submit(fn, *args, **kwargs): + scheduled.append((fn, args, kwargs)) + return Future() + + cluster.executor.submit = delayed_submit + order = [] + cluster.profile_manager.on_down.side_effect = lambda _: order.append('down') + cluster.profile_manager.on_up.side_effect = lambda _: order.append('up') + try: + assert cluster.on_down(host, False) + assert order == [] + cluster.on_up(host) + assert order == ['down', 'up'] + for fn, args, kwargs in scheduled: + fn(*args, **kwargs) + assert order == ['down', 'up'] + assert host.is_up + finally: + cluster.executor.submit = original_submit + + +def test_rejected_executor_handoff_removes_queued_down(topology): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + cluster._discount_down_events = False + original_submit = cluster.executor.submit + cluster.executor.submit = Mock(side_effect=RuntimeError('executor stopped')) + try: + assert not cluster.on_down(host, False) + assert not host._currently_handling_node_down + cluster._lifecycle.drain() + cluster.profile_manager.on_down.assert_not_called() + finally: + cluster.executor.submit = original_submit