pool: release stopped host reconnection handlers

CHANGELOG.rst

@@ -29,6 +29,13 @@ Features
   statements skip re-sending result metadata on EXECUTE, and the driver automatically
   refreshes cached metadata when the server detects a schema change (DRIVER-153)
 
+Bug Fixes
+---------
+* A host reconnection handler now releases the host's reconnection slot when
+  authentication fails or its retry schedule is exhausted. A later DOWN event can
+  therefore start a new handler after credentials recover or another reconnection
+  opportunity appears, instead of treating the stopped handler as an active one (#1026).
+
 Others
 ------
 * ``DCAwareRoundRobinPolicy.local_dc`` is now read-only. It is set by the constructor,

cassandra/cluster.py

@@ -194,6 +194,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")
 
@@ -1987,6 +1988,9 @@ def on_up(self, host):
         return futures
 
     def _start_reconnector(self, host, is_host_addition):
+        if self.metadata.get_host_by_host_id(host.host_id) is not host:
+            return
+
         if self.profile_manager.distance(host) == HostDistance.IGNORED:
             return
 
@@ -2012,15 +2016,22 @@ def _start_reconnector(self, host, is_host_addition):
 
     @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)
+        try:
+            if self.metadata.get_host_by_host_id(host.host_id) is not host:
+                return
 
-        for listener in self.listeners:
-            listener.on_down(host)
+            self.profile_manager.on_down(host)
+            self.control_connection.on_down(host)
+            for session in tuple(self.sessions):
+                session.on_down(host)
 
-        self._start_reconnector(host, is_host_addition)
+            for listener in self.listeners:
+                listener.on_down(host)
+
+            self._start_reconnector(host, is_host_addition)
+        finally:
+            with host.lock:
+                host._currently_handling_node_down = False
 
     def on_down(self, host, is_host_addition, expect_host_to_be_down=False):
         """
@@ -2029,7 +2040,13 @@ 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
 
+        if self.metadata.get_host_by_host_id(host.host_id) is not host:
+            return
+
         with host.lock:
+            if self.metadata.get_host_by_host_id(host.host_id) is not host:
+                return
+
             was_up = host.is_up
 
             # ignore down signals if we have open pools to the host
@@ -2045,11 +2062,24 @@ def on_down(self, host, is_host_addition, expect_host_to_be_down=False):
                     return
 
             host.set_down()
-            if (not was_up and not expect_host_to_be_down) or host.is_currently_reconnecting():
+            if (host.is_currently_reconnecting() or
+                    host._currently_handling_node_down):
                 return
+
+            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.
+                self._start_reconnector(host, is_host_addition)
+                return
+
+            host._currently_handling_node_down = True
         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)
+        if future is None:
+            with host.lock:
+                host._currently_handling_node_down = False
 
     def on_add(self, host, refresh_nodes=True):
         if self.is_shutdown:
@@ -2121,7 +2151,7 @@ def _finalize_add(self, host, set_up=True):
         for session in tuple(self.sessions):
             session.update_created_pools()
 
-    def on_remove(self, host):
+    def on_remove(self, host, refresh_nodes=True):
         if self.is_shutdown:
             return
 
@@ -2132,7 +2162,7 @@ def on_remove(self, host):
             session.on_remove(host)
         for listener in self.listeners:
             listener.on_remove(host)
-        self.control_connection.on_remove(host)
+        self.control_connection.on_remove(host, refresh_nodes=refresh_nodes)
 
         reconnection_handler = host.get_and_set_reconnection_handler(None)
         if reconnection_handler:
@@ -2162,14 +2192,31 @@ def add_host(self, endpoint, datacenter=None, rack=None, signal=True, refresh_no
 
         return host, new
 
-    def remove_host(self, host):
+    def remove_host(self, host, refresh_nodes=True):
         """
         Called when the control connection observes that a node has left the
         ring.  Intended for internal use only.
         """
         if host and self.metadata.remove_host(host):
             log.info("Cassandra host %s removed", host)
-            self.on_remove(host)
+            self.on_remove(host, refresh_nodes=refresh_nodes)
+
+    def remove_host_by_host_id(self, host_id, endpoint=None,
+                               refresh_nodes=True):
+        """Remove the host stored under a specific metadata key."""
+        host = self.metadata.get_host_by_host_id(host_id)
+        if not host or not self.metadata.remove_host_by_host_id(
+                host_id, endpoint):
+            return
+
+        # A refresh can reindex the same Host under a new host ID before
+        # cleaning up its stale old key.  In that case only the alias was
+        # removed; the Host itself is still part of the cluster.
+        if self.metadata.get_host_by_host_id(host.host_id) is host:
+            return
+
+        log.info("Cassandra host %s removed", host)
+        self.on_remove(host, refresh_nodes=refresh_nodes)
 
     def register_listener(self, listener):
         """
@@ -4232,7 +4279,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)
-                self._cluster.metadata.remove_host_by_host_id(old_host_id, old_host.endpoint)
+                self._cluster.remove_host_by_host_id(
+                    old_host_id, old_host.endpoint, refresh_nodes=False)
 
         log.debug("[control connection] Finished fetching ring info")
         if partitioner and should_rebuild_token_map:
@@ -4649,13 +4697,13 @@ 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):
+    def on_remove(self, host, refresh_nodes=True):
         c = self._connection
         if self._connection_matches_host(c, host):
             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):

cassandra/pool.py

@@ -161,6 +161,7 @@ class Host(object):
     lock = None
 
     _currently_handling_node_up = False
+    _currently_handling_node_down = False
 
     sharding_info = None
 
@@ -355,6 +356,18 @@ def __init__(self, host, connection_factory, is_host_addition, on_add, on_up, *a
         self.host = host
         self.connection_factory = connection_factory
 
+    def start(self):
+        try:
+            _ReconnectionHandler.start(self)
+        except StopIteration:
+            # An empty schedule means this handler will never run.
+            self._release_slot()
+
+    def _release_slot(self):
+        with self.host.lock:
+            if self.host._reconnection_handler is self:
+                self.host._reconnection_handler = None
+
     def try_reconnect(self):
         return self.connection_factory()
 
@@ -367,12 +380,19 @@ def on_reconnection(self, connection):
 
     def on_exception(self, exc, next_delay):
         if isinstance(exc, AuthenticationFailed):
-            return False
+            keep_retrying = False
         else:
             log.warning("Error attempting to reconnect to %s, scheduling retry in %s seconds: %s",
                         self.host, next_delay, exc)
             log.debug("Reconnection error details", exc_info=True)
-            return True
+            keep_retrying = True
+
+        if not keep_retrying or next_delay is None:
+            # This handler will never run again. Release the slot it occupies,
+            # or a later DOWN event will mistake it for a live reconnector.
+            self._release_slot()
+
+        return keep_retrying
 
 
 class HostConnection(object):