pool: release stopped host reconnection handlers

CHANGELOG.rst

@@ -55,6 +55,10 @@ Bug Fixes
   processed before ``send_msg()`` returns. On a later timeout the driver then
   orphaned the wrong stream id and left the real request in the connection's
   request map (#1013).
+* 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
 ------

cassandra/cluster.py

@@ -195,6 +195,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")
 
@@ -2067,7 +2068,9 @@ 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()
+            with host.lock:
+                host.set_up()
+                host._pending_host_addition = False
             for listener in self.listeners:
                 listener.on_up(host)
         finally:
@@ -2145,6 +2148,7 @@ def on_up(self, host):
             if not have_future:
                 with host.lock:
                     host.set_up()
+                    host._pending_host_addition = False
                     host._currently_handling_node_up = False
 
         # for testing purposes
@@ -2161,12 +2165,15 @@ def _start_reconnector(self, host, is_host_addition):
         # of the current Cluster attributes to create new Connections with
         conn_factory = self._make_connection_factory(host)
 
-        reconnector = _HostReconnectionHandler(
-            host, conn_factory, is_host_addition, self.on_add, self.on_up,
-            self.scheduler, schedule, host.get_and_set_reconnection_handler,
-            new_handler=None)
-
-        old_reconnector = host.get_and_set_reconnection_handler(reconnector)
+        with host.lock:
+            if is_host_addition:
+                host._pending_host_addition = True
+            is_host_addition = host._pending_host_addition
+            reconnector = _HostReconnectionHandler(
+                host, conn_factory, is_host_addition, self.on_add, self.on_up,
+                self.scheduler, schedule,
+                host.get_and_set_reconnection_handler, new_handler=None)
+            old_reconnector = host.get_and_set_reconnection_handler(reconnector)
         if old_reconnector:
             log.debug("Old host reconnector found for %s, cancelling", host)
             old_reconnector.cancel()
@@ -2175,16 +2182,46 @@ 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, is_host_addition,
+                                     down_event_generation):
+        try:
+            with host.lock:
+                if down_event_generation != host._down_event_generation:
+                    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)
+
+            for listener in self.listeners:
+                listener.on_down(host)
 
-        self._start_reconnector(host, is_host_addition)
+            self._start_reconnector(host, is_host_addition)
+        finally:
+            with host.lock:
+                if down_event_generation == host._down_event_generation:
+                    host._currently_handling_node_down = False
+
+    @run_in_executor
+    def _restart_reconnector(self, host, is_host_addition,
+                             down_event_generation):
+        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()):
+                    self._start_reconnector(host, is_host_addition)
+        finally:
+            with host.lock:
+                if down_event_generation == host._down_event_generation:
+                    host._currently_handling_node_down = False
 
     def on_down(self, host, is_host_addition, expect_host_to_be_down=False):
         """
@@ -2193,6 +2230,7 @@ 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
 
+        restart_reconnector = False
         with host.lock:
             was_up = host.is_up
 
@@ -2209,16 +2247,46 @@ 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 (not was_up and
+                    (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.
+                restart_reconnector = True
+
+            host._currently_handling_node_down = True
+            host._down_event_generation += 1
+            down_event_generation = host._down_event_generation
+
+        if restart_reconnector:
+            future = self._restart_reconnector(
+                host, is_host_addition, down_event_generation)
+            if future is None:
+                with host.lock:
+                    if down_event_generation == host._down_event_generation:
+                        host._currently_handling_node_down = False
+            return
+
         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, down_event_generation)
+        if future is None:
+            with host.lock:
+                if down_event_generation == host._down_event_generation:
+                    host._currently_handling_node_down = False
 
     def on_add(self, host, refresh_nodes=True):
         if self.is_shutdown:
             return
 
+        with host.lock:
+            host._pending_host_addition = True
+
         log.debug("Handling new host %r and notifying listeners", host)
 
         self.profile_manager.on_add(host)
@@ -2275,8 +2343,10 @@ def future_completed(future):
             self._finalize_add(host)
 
     def _finalize_add(self, host, set_up=True):
-        if set_up:
-            host.set_up()
+        with host.lock:
+            if set_up:
+                host.set_up()
+            host._pending_host_addition = False
 
         for listener in self.listeners:
             listener.on_add(host)

cassandra/pool.py

@@ -161,6 +161,9 @@ class Host(object):
     lock = None
 
     _currently_handling_node_up = False
+    _currently_handling_node_down = False
+    _down_event_generation = 0
+    _pending_host_addition = False
 
     sharding_info = None
 
@@ -355,6 +358,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 +382,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):