cluster: replay up events queued during down handling

Previously, an on_up() signal that arrived while on_down_potentially_blocking() was still notifying policies, sessions, and listeners could be dropped. That left the host down even though the latest observed state was up.

Queue a single pending up replay while down handling is active, then replay it after the down path clears. The replay is guarded by _node_down_event_revision so a newer down or remove event invalidates the older up instead of letting it mark the host up later.

Reserve down handling before submitting executor work so an up arriving between submit and worker execution is queued deterministically. Use handling revisions to prevent stale async callbacks from clearing newer in-flight work.

Duplicate down signals still avoid duplicate down notification and reconnector work. A later down only invalidates pending up replay. Unknown hosts from auth failures remain no-op unless the caller explicitly passes expect_host_to_be_down.

Author: dkropachev

cassandra/cluster.py

@@ -29,7 +29,7 @@
 from itertools import groupby, count, chain
 import json
 import logging
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, NamedTuple, Optional, Union
 from warnings import warn
 from random import random
 import re
@@ -234,6 +234,7 @@ def new_f(self, *args, **kwargs):
         try:
             future = self.executor.submit(f, self, *args, **kwargs)
             future.add_done_callback(_future_completed)
+            return future
         except Exception:
             log.exception("Failed to submit task to executor")
 
@@ -251,6 +252,15 @@ def _discard_cluster_shutdown(cluster):
     _clusters_for_shutdown.discard(cluster)
 
 
+class _StaleDownProtection(NamedTuple):
+    # Snapshot captured when a down event is scheduled, so delayed executor
+    # handling can ignore work that was superseded by newer host state.
+    # down_event_revision is None only for unreserved direct worker calls;
+    # then the worker must not clear a handling slot owned by a newer event.
+    host_state_revision: int
+    down_event_revision: Optional[int]
+
+
 def _shutdown_clusters():
     clusters = _clusters_for_shutdown.copy()  # copy because shutdown modifies the global set "discard"
     for cluster in clusters:
@@ -1870,7 +1880,61 @@ def _cleanup_failed_on_up_handling(self, host):
 
         self._start_reconnector(host, is_host_addition=False)
 
-    def _on_up_future_completed(self, host, futures, results, lock, finished_future):
+    def _up_handling_was_superseded(self, host, up_handling_revision):
+        return (getattr(host, "_node_up_handling_revision", None) != up_handling_revision or
+                host._node_down_event_revision != up_handling_revision)
+
+    def _clear_up_handling(self, host, up_handling_revision=None):
+        if (up_handling_revision is not None and
+                getattr(host, "_node_up_handling_revision", None) != up_handling_revision):
+            return False
+        host._currently_handling_node_up = False
+        host._node_up_handling_revision = None
+        return True
+
+    def _cleanup_superseded_up_handling(self, host):
+        for session in tuple(self.sessions):
+            session.remove_pool(host)
+
+    def _pop_pending_node_up_if_ready(self, host):
+        if not host._pending_node_up:
+            return None
+        if host.is_up:
+            host._pending_node_up = False
+            host._pending_node_up_revision = None
+            return None
+        if host._currently_handling_node_up or host._currently_handling_node_down:
+            return None
+
+        pending_up_revision = host._pending_node_up_revision
+        # Leave the pending marker in place until on_up() reacquires host.lock so
+        # a newer down signal can still invalidate this replay.
+        return pending_up_revision
+
+    def _handle_pending_node_up(self, host, pending_up_revision):
+        if pending_up_revision is not None:
+            log.debug("Handling queued up status of node %s", host)
+            self.on_up(host, expected_down_event_revision=pending_up_revision)
+
+    def _clear_down_handling(self, host, down_event_revision=None):
+        if (down_event_revision is not None and
+                getattr(host, "_node_down_handling_revision", None) != down_event_revision):
+            return False
+        host._currently_handling_node_down = False
+        host._node_down_handling_revision = None
+        return True
+
+    def _finish_superseded_up_handling(self, host, up_handling_revision):
+        self._cleanup_superseded_up_handling(host)
+
+        pending_up_revision = None
+        with host.lock:
+            if self._clear_up_handling(host, up_handling_revision):
+                pending_up_revision = self._pop_pending_node_up_if_ready(host)
+
+        self._handle_pending_node_up(host, pending_up_revision)
+
+    def _on_up_future_completed(self, host, up_handling_revision, futures, results, lock, finished_future):
         with lock:
             futures.discard(finished_future)
 
@@ -1896,18 +1960,32 @@ def _on_up_future_completed(self, host, futures, results, lock, finished_future)
 
             log.info("Connection pools established for node %s", host)
             # mark the host as up and notify all listeners
-            host.set_up()
+            superseded = False
+            with host.lock:
+                if self._up_handling_was_superseded(host, up_handling_revision):
+                    log.debug("Ignoring superseded up handling for node %s", host)
+                    superseded = True
+                else:
+                    host.set_up()
+                    self._clear_up_handling(host, up_handling_revision)
+            if superseded:
+                self._finish_superseded_up_handling(host, up_handling_revision)
+                return
             for listener in self.listeners:
                 listener.on_up(host)
         finally:
+            pending_up_revision = None
             with host.lock:
-                host._currently_handling_node_up = False
+                if self._clear_up_handling(host, up_handling_revision):
+                    pending_up_revision = self._pop_pending_node_up_if_ready(host)
+            self._handle_pending_node_up(host, pending_up_revision)
 
         # see if there are any pools to add or remove now that the host is marked up
         for session in tuple(self.sessions):
             session.update_created_pools()
+        return
 
-    def on_up(self, host):
+    def on_up(self, host, expected_down_event_revision=None):
         """
         Intended for internal use only.
         """
@@ -1916,14 +1994,38 @@ def on_up(self, host):
 
         log.debug("Waiting to acquire lock for handling up status of node %s", host)
         with host.lock:
+            if (expected_down_event_revision is not None and
+                    host._node_down_event_revision != expected_down_event_revision):
+                log.debug("Ignoring stale queued up handling for node %s", host)
+                return
+
+            if host._currently_handling_node_down:
+                log.debug("Down status is being handled for node %s; queueing up handling", host)
+                host._pending_node_up = True
+                host._pending_node_up_revision = host._node_down_event_revision
+                return
+
             if host._currently_handling_node_up:
-                log.debug("Another thread is already handling up status of node %s", host)
+                up_handling_revision = getattr(host, "_node_up_handling_revision", None)
+                if self._up_handling_was_superseded(host, up_handling_revision):
+                    log.debug("Superseded up handling is still finishing for node %s; "
+                              "queueing up handling", host)
+                    host._pending_node_up = True
+                    host._pending_node_up_revision = host._node_down_event_revision
+                else:
+                    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)
+                host._pending_node_up = False
+                host._pending_node_up_revision = None
                 return
 
+            host._pending_node_up = False
+            host._pending_node_up_revision = None
+            up_handling_revision = host._node_down_event_revision
+            host._node_up_handling_revision = up_handling_revision
             host._currently_handling_node_up = True
         log.debug("Starting to handle up status of node %s", host)
 
@@ -1953,7 +2055,8 @@ def on_up(self, host):
             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)
+            callback = partial(self._on_up_future_completed, host, up_handling_revision,
+                               futures, futures_results, futures_lock)
             for session in tuple(self.sessions):
                 future = session.add_or_renew_pool(host, is_host_addition=False)
                 if future is not None:
@@ -1967,14 +2070,24 @@ def on_up(self, host):
 
             self._cleanup_failed_on_up_handling(host)
 
+            pending_up_revision = None
             with host.lock:
-                host._currently_handling_node_up = False
+                if self._clear_up_handling(host, up_handling_revision):
+                    pending_up_revision = self._pop_pending_node_up_if_ready(host)
+            self._handle_pending_node_up(host, pending_up_revision)
             raise
         else:
             if not have_future:
+                superseded = False
                 with host.lock:
-                    host.set_up()
-                    host._currently_handling_node_up = False
+                    if self._up_handling_was_superseded(host, up_handling_revision):
+                        log.debug("Ignoring superseded up handling for node %s", host)
+                        superseded = True
+                    else:
+                        host.set_up()
+                        self._clear_up_handling(host, up_handling_revision)
+                if superseded:
+                    self._finish_superseded_up_handling(host, up_handling_revision)
 
         # for testing purposes
         return futures
@@ -2004,16 +2117,65 @@ def _start_reconnector(self, host, is_host_addition):
         reconnector.start()
 
     @run_in_executor
-    def on_down_potentially_blocking(self, host, is_host_addition):
-        self.profile_manager.on_down(host)
-        self.control_connection.on_down(host)
-        for session in tuple(self.sessions):
-            session.on_down(host)
+    def on_down_potentially_blocking(
+            self, host: Host, is_host_addition: bool,
+            stale_down_protection: Optional[_StaleDownProtection] = None) -> Any:
+        handle_stale_down_event = False
+        pending_up_revision = None
+        down_event_revision = None
+        if stale_down_protection is not None:
+            down_event_revision = stale_down_protection.down_event_revision
+        with host.lock:
+            down_handling_revision = getattr(host, "_node_down_handling_revision", None)
+            owns_reserved_down_handling = (
+                down_event_revision is not None and
+                host._currently_handling_node_down and
+                down_handling_revision == down_event_revision
+            )
+            node_up_handling_revision = getattr(host, "_node_up_handling_revision", None)
+            stale_down_event = (
+                host.is_up or
+                (host._currently_handling_node_up and
+                 (down_event_revision is None or
+                  node_up_handling_revision is None or
+                  down_event_revision <= node_up_handling_revision)) or
+                (stale_down_protection is not None and
+                 host._state_revision != stale_down_protection.host_state_revision)
+            )
+            if stale_down_event:
+                log.debug("Ignoring stale down handling for host %s", host)
+                if owns_reserved_down_handling and self._clear_down_handling(host, down_event_revision):
+                    pending_up_revision = self._pop_pending_node_up_if_ready(host)
+                handle_stale_down_event = True
+            elif host._currently_handling_node_down:
+                if not owns_reserved_down_handling:
+                    log.debug("Another thread is already handling down status of node %s", host)
+                    return
+            else:
+                host._currently_handling_node_down = True
+                host._node_down_handling_revision = down_event_revision
 
-        for listener in self.listeners:
-            listener.on_down(host)
+        if handle_stale_down_event:
+            self._handle_pending_node_up(host, pending_up_revision)
+            return
 
-        self._start_reconnector(host, is_host_addition)
+        try:
+            self.profile_manager.on_down(host)
+            self.control_connection.on_down(host)
+            for session in tuple(self.sessions):
+                session.on_down(host)
+
+            for listener in self.listeners:
+                listener.on_down(host)
+
+            self._start_reconnector(host, is_host_addition)
+        finally:
+            pending_up_revision = None
+            with host.lock:
+                if self._clear_down_handling(host, down_event_revision):
+                    pending_up_revision = self._pop_pending_node_up_if_ready(host)
+
+            self._handle_pending_node_up(host, pending_up_revision)
 
     def on_down(self, host, is_host_addition, expect_host_to_be_down=False):
         """
@@ -2037,12 +2199,40 @@ def on_down(self, host, is_host_addition, expect_host_to_be_down=False):
                 if connected:
                     return
 
+            if not expect_host_to_be_down:
+                if was_up is False:
+                    if host._pending_node_up:
+                        host._node_down_event_revision += 1
+                        host._pending_node_up = False
+                        host._pending_node_up_revision = None
+                        host.set_down()
+                    return
+                if was_up is None:
+                    return
+
+            host._node_down_event_revision += 1
+            host._pending_node_up = False
+            host._pending_node_up_revision = None
             host.set_down()
-            if (not was_up and not expect_host_to_be_down) or host.is_currently_reconnecting():
+            stale_down_protection = _StaleDownProtection(
+                host_state_revision=host._state_revision,
+                down_event_revision=host._node_down_event_revision)
+            if host.is_currently_reconnecting():
+                return
+            if host._currently_handling_node_down:
                 return
+            host._currently_handling_node_down = True
+            host._node_down_handling_revision = stale_down_protection.down_event_revision
         log.warning("Host %s has been marked down", host)
 
-        self.on_down_potentially_blocking(host, is_host_addition)
+        future = self.on_down_potentially_blocking(
+            host, is_host_addition, stale_down_protection)
+        if future is None:
+            pending_up_revision = None
+            with host.lock:
+                if self._clear_down_handling(host, stale_down_protection.down_event_revision):
+                    pending_up_revision = self._pop_pending_node_up_if_ready(host)
+            self._handle_pending_node_up(host, pending_up_revision)
 
     def on_add(self, host, refresh_nodes=True):
         if self.is_shutdown:
@@ -2119,7 +2309,11 @@ def on_remove(self, host):
             return
 
         log.debug("[cluster] Removing host %s", host)
-        host.set_down()
+        with host.lock:
+            host._node_down_event_revision += 1
+            host._pending_node_up = False
+            host._pending_node_up_revision = None
+            host.set_down()
         self.profile_manager.on_remove(host)
         for session in tuple(self.sessions):
             session.on_remove(host)