topology: separate publication from lifecycle notifications

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:

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()
 

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):

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.