diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index e3eaa2c24e..317a94e303 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -59,6 +59,42 @@ Bug Fixes
   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).
+* A defunct control connection is no longer left unreconnected when the cluster does not
+  run DOWN handling for its host (#847). ``ControlConnection._signal_error()`` treated the
+  conviction policy accepting a failure as a guarantee that a DOWN callback would
+  reconnect it, but the two are not the same: ``Cluster.on_down()`` deliberately skips
+  DOWN handling when a session pool to the host is still open, when the host is already
+  down or already reconnecting, and when pool creation is disabled -- and the default
+  ``SimpleConvictionPolicy`` rejects the conviction outright for ``OperationTimedOut``.
+  In all of those cases the control connection stayed defunct with nothing scheduled to
+  replace it. ``Cluster.on_down()`` and ``Cluster.signal_connection_failure()`` now return
+  whether DOWN handling was actually dispatched, and the control connection reconnects
+  directly whenever it was not. This applies uniformly to TCP, Unix socket,
+  alternate-route and stable host-ID connections, and does not duplicate the reconnect
+  that an accepted DOWN transition already performs. Two related cases are fixed with
+  it: a control connection whose reconnection attempts are already backing off no longer
+  has that schedule cancelled and restarted from its initial delay by every further
+  error, and a reconnection handler that has stopped for good -- once its retry schedule
+  is exhausted -- now releases the slot it occupies, so a later error starts a fresh
+  reconnection instead of mistaking the dead handler for one still retrying. The slot is
+  likewise released once a connection is installed, so a handler whose backoff outlived
+  the reconnection that succeeded without it no longer blocks the next one. Every release
+  is identity-checked, so a handler that stops can only ever clear itself: it cannot evict
+  a replacement another thread installed while it was handing its connection over, which
+  would have left that replacement retrying where nothing could find it. An attempt that
+  fails while an overlapping one succeeds no longer parks a handler in the slot the
+  successful attempt emptied, which would have blocked reconnection for the length of its
+  backoff and then replaced a healthy control connection. Once retry handling begins, its
+  cadence follows ``reconnection_policy`` up to ``max_reconnection_delay`` instead of
+  being restarted by each ``idle_heartbeat_interval``. Operators that need a shorter
+  recovery bound should configure a lower maximum delay. A finite schedule gives up after
+  its configured attempts, and recurring heartbeat returns for the same failed connection
+  do not re-arm it; continued automatic recovery requires an infinite schedule.
+* The control connection is no longer closed immediately after a reconnection handler
+  restores it. ``_ReconnectionHandler.run()`` closed the connection it had just opened,
+  which is right for the host handler that only uses it to probe the host, but left the
+  control connection dead the moment its backoff finally succeeded -- until a heartbeat
+  noticed, or forever with ``idle_heartbeat_interval=0``.
 
 Others
 ------
diff --git a/cassandra/cluster.py b/cassandra/cluster.py
index f8859a5659..a7b4d620ac 100644
--- a/cassandra/cluster.py
+++ b/cassandra/cluster.py
@@ -185,6 +185,10 @@ def _future_completed(future):
 def run_in_executor(f):
     """
     A decorator to run the given method in the ThreadPoolExecutor.
+
+    The wrapper returns the submitted Future, or None when the cluster is
+    shutting down or the executor rejected the submission, in which case the
+    wrapped method never runs.
     """
 
     @wraps(f)
@@ -2189,13 +2193,32 @@ def on_down_potentially_blocking(self, host, is_host_addition,
                 if down_event_generation != host._down_event_generation:
                     return
 
-            self.profile_manager.on_down(host)
-            self.control_connection.on_down(host)
+            # The load balancing policies go first, so that the query plan the
+            # control connection's reconnect walks no longer offers the host
+            # that just went down; otherwise the reconnect submitted below can
+            # start before the policies have dropped it and burn a
+            # connect_timeout on a dead node. Both calls are guarded because
+            # on_down() reports successful dispatch to callers that rely on a
+            # reconnection being started.
+            try:
+                self.profile_manager.on_down(host)
+            except Exception:
+                log.exception("Error in load balancing policy down handler for host %s", host)
+            try:
+                self.control_connection.on_down(host)
+            except Exception:
+                log.exception("Error in control connection down handler for host %s", host)
             for session in tuple(self.sessions):
-                session.on_down(host)
+                try:
+                    session.on_down(host)
+                except Exception:
+                    log.exception("Error marking host %s down in session", host)
 
             for listener in self.listeners:
-                listener.on_down(host)
+                try:
+                    listener.on_down(host)
+                except Exception:
+                    log.exception("Error in host state listener down handler for host %s", host)
 
             self._start_reconnector(host, is_host_addition)
         finally:
@@ -2226,9 +2249,11 @@ def _restart_reconnector(self, host, is_host_addition,
     def on_down(self, host, is_host_addition, expect_host_to_be_down=False):
         """
         Intended for internal use only.
+
+        Returns whether DOWN handling was dispatched.
         """
         if self.is_shutdown or self.allow_control_connection_query_fallback == ControlConnectionQueryFallback.SkipPoolCreation:
-            return
+            return False
 
         restart_reconnector = False
         with host.lock:
@@ -2244,13 +2269,13 @@ def on_down(self, host, is_host_addition, expect_host_to_be_down=False):
                     if pool_state:
                         connected |= pool_state['open_count'] > 0
                 if connected:
-                    return
+                    return False
 
             host.set_down()
             if (not was_up and
                     (host.is_currently_reconnecting() or
                      host._currently_handling_node_down)):
-                return
+                return False
 
             if not was_up and not expect_host_to_be_down:
                 # A terminal failure may have released this down host's old
@@ -2269,7 +2294,9 @@ def on_down(self, host, is_host_addition, expect_host_to_be_down=False):
                 with host.lock:
                     if down_event_generation == host._down_event_generation:
                         host._currently_handling_node_down = False
-            return
+            # 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)
 
@@ -2279,6 +2306,7 @@ def on_down(self, host, is_host_addition, expect_host_to_be_down=False):
             with host.lock:
                 if down_event_generation == host._down_event_generation:
                     host._currently_handling_node_down = False
+        return future is not None
 
     def on_add(self, host, refresh_nodes=True):
         if self.is_shutdown:
@@ -2373,10 +2401,11 @@ def on_remove(self, host):
             reconnection_handler.cancel()
 
     def signal_connection_failure(self, host, connection_exc, is_host_addition, expect_host_to_be_down=False):
+        """Return whether the failure caused DOWN handling to be dispatched."""
         is_down = host.signal_connection_failure(connection_exc)
-        if is_down:
-            self.on_down(host, is_host_addition, expect_host_to_be_down)
-        return is_down
+        if not is_down:
+            return False
+        return self.on_down(host, is_host_addition, expect_host_to_be_down)
 
     def add_host(self, endpoint, datacenter=None, rack=None, signal=True, refresh_nodes=True, host_id=None):
         """
@@ -3971,23 +4000,132 @@ class _ControlReconnectionHandler(_ReconnectionHandler):
     Internal
     """
 
-    def __init__(self, control_connection, *args, **kwargs):
-        _ReconnectionHandler.__init__(self, *args, **kwargs)
+    # on_reconnection() installs the connection as the control connection,
+    # so it must outlive the handler run that opened it.
+    _keeps_connection = True
+
+    def __init__(self, control_connection, scheduler, schedule):
+        # The run-completed callback releases this handler's own slot. It is
+        # identity-checked, so a run that finishes after another thread has
+        # installed a replacement leaves that replacement alone. Anything that
+        # evicts a handler must also cancel it, or the evicted one keeps
+        # retrying where nothing can find it.
+        _ReconnectionHandler.__init__(self, scheduler, schedule, self._release)
         self.control_connection = weakref.proxy(control_connection)
+        # Remember a failed connection that caused this retry run. If the
+        # finite schedule is exhausted, the heartbeat will keep returning this
+        # same object and must not start the schedule over from the beginning.
+        # A healthy connection may also be replaced after a topology event;
+        # exhausting that attempt must not suppress recovery if the connection
+        # fails later for an independent reason.
+        connection = control_connection._connection
+        self._failed_connection = connection \
+            if connection is not None and (
+                connection.is_defunct or connection.is_closed) else None
+        # A reconnect trigger that arrives while an attempt is actually
+        # running is normally covered by that attempt. If this turns out to be
+        # the final failed attempt, though, there is no later retry to own the
+        # trigger, so _release() must hand it back to reconnect().
+        self._is_running = False
+        self._run_generation = 0
+        self._reconnect_requested = False
+        # A proactive handler may be waiting while its still-healthy control
+        # connection fails. Its remaining retries own the immediate recovery,
+        # but the failure must receive a fresh schedule if those retries are
+        # exhausted. Unlike a trigger received during a running attempt, this
+        # state therefore survives every non-final handler run.
+        self._new_failure_detected = False
+
+    def run(self):
+        try:
+            control_connection = self.control_connection
+            with control_connection._reconnection_lock:
+                self._run_generation += 1
+                run_generation = self._run_generation
+                self._is_running = True
+        except ReferenceError:
+            return _ReconnectionHandler.run(self)
+
+        try:
+            _ReconnectionHandler.run(self)
+        finally:
+            try:
+                with control_connection._reconnection_lock:
+                    # A zero-delay retry may already be running on another
+                    # worker. Only the invocation that most recently claimed
+                    # the state may clear it.
+                    if self._run_generation == run_generation:
+                        self._is_running = False
+            except ReferenceError:
+                pass
 
     def try_reconnect(self):
         return self.control_connection._reconnect_internal()
 
     def on_reconnection(self, connection):
-        self.control_connection._set_new_connection(connection)
+        try:
+            # Resolving the attribute is what dereferences the weak proxy, so
+            # it is bound here rather than called directly: a ReferenceError
+            # raised from inside _set_new_connection() would mean the
+            # connection was already adopted, and must not be closed.
+            set_new_connection = self.control_connection._set_new_connection
+        except ReferenceError:
+            # The ControlConnection was collected while we were retrying. The
+            # handler has already marked the connection as handed off, so
+            # nothing else would ever close it.
+            connection.close()
+            return
+        set_new_connection(connection)
 
     def on_exception(self, exc, next_delay):
-        # TODO only overridden to add logging, so add logging
-        if isinstance(exc, AuthenticationFailed):
-            return False
+        # Every reason to stop retrying is worth retrying here: an attempt
+        # covers the whole query plan, so an authentication failure is a
+        # failure of the host it happened to reach, not of the cluster.
+        log.debug("Error trying to reconnect control connection: %r", exc)
+
+        if next_delay is None:
+            # The schedule is exhausted, so this handler will never run again.
+            # Release the slot it occupies while remembering the connection
+            # whose retries were exhausted. A later explicit error may start a
+            # fresh run, but heartbeats for this same defunct connection must
+            # not reset the finite schedule on every interval.
+            self._release(exhausted=True)
         else:
-            log.debug("Error trying to reconnect control connection: %r", exc)
-            return True
+            # This failure already has another scheduled attempt to own any
+            # trigger that arrived while it was running.
+            try:
+                control_connection = self.control_connection
+                with control_connection._reconnection_lock:
+                    self._is_running = False
+                    self._reconnect_requested = False
+            except ReferenceError:
+                pass
+
+        return True
+
+    def _release(self, exhausted=False):
+        reconnect_requested = False
+        try:
+            control_connection = self.control_connection
+            with control_connection._reconnection_lock:
+                if control_connection._reconnection_handler is self:
+                    control_connection._reconnection_handler = None
+                    reconnect_requested = exhausted and (
+                        self._reconnect_requested or
+                        self._new_failure_detected)
+                    self._reconnect_requested = False
+                    self._new_failure_detected = False
+                    if (exhausted and not reconnect_requested and
+                            self._failed_connection is not None and
+                            control_connection._connection is
+                            self._failed_connection):
+                        control_connection._reconnection_exhausted_connection = \
+                            self._failed_connection
+        except ReferenceError:
+            pass  # our weak reference to the ControlConnection is no good
+
+        if reconnect_requested:
+            control_connection.reconnect()
 
 
 def _watch_callback(obj_weakref, method_name, *args, **kwargs):
@@ -4084,6 +4222,24 @@ def __init__(self, cluster, timeout,
 
         self._reconnection_handler = None
         self._reconnection_lock = RLock()
+        # A finite retry schedule may be exhausted while the heartbeat still
+        # owns and returns the same defunct connection. Keep that identity so
+        # heartbeat passes do not create a fresh schedule indefinitely.
+        self._reconnection_exhausted_connection = None
+        self._reconnect_pending = False
+        # Set when reconnect() is called while an attempt is already pending.
+        # If that attempt fails without installing a retry handler, its final
+        # action consumes this flag by queuing one follow-up attempt.
+        self._reconnect_requested = False
+        # Bumped every time _reconnect_pending is raised. An attempt clears the
+        # flag only while it still owns it: _set_new_connection() drops the flag
+        # as the connection goes in, so a newer reconnect() can claim it before
+        # the attempt that installed the connection reaches its finally clause,
+        # and clearing it there would strand the attempt that newer flag owns.
+        self._reconnect_pending_seq = 0
+        # Bumped every time a connection is installed. _reconnect() compares it
+        # across its attempt to tell whether another attempt got there first.
+        self._connection_generation = 0
 
         self._event_schedule_times = {}
 
@@ -4113,20 +4269,61 @@ def connect(self):
         self._protocol_version = self._cluster.protocol_version
         self._set_new_connection(self._reconnect_internal())
 
-        self._cluster.metadata.dbaas = self._connection._product_type == dscloud.DATASTAX_CLOUD_PRODUCT_TYPE
+        # _set_new_connection declines to install anything once shutdown() has
+        # run, so there may be no connection to ask.
+        if self._connection:
+            self._cluster.metadata.dbaas = self._connection._product_type == dscloud.DATASTAX_CLOUD_PRODUCT_TYPE
 
     def _set_new_connection(self, conn):
         """
-        Replace existing connection (if there is one) and close it.
+        Adopt `conn` as the control connection, closing the one it replaces.
+
+        Also ends any reconnection in progress: a handler parked in the slot is
+        cancelled and cleared, because it would otherwise be mistaken for one
+        still retrying and its next attempt would replace this connection.
+
+        If the ControlConnection has already been shut down, `conn` is not
+        installed at all -- nothing would ever use it or close it on our
+        behalf, so it is closed here instead.
         """
-        with self._lock:
-            old = self._connection
-            self._connection = conn
+        # Whatever put this connection in place, the reconnection is over. A
+        # handler left parked in the slot would be mistaken for one still
+        # retrying, and its next attempt would replace this connection.
+        #
+        # _reconnection_lock is held across the install so that the generation
+        # bump is visible to any _reconnect() that is about to decide whether
+        # another attempt beat it; releasing it first leaves a window where a
+        # failing attempt reads the old generation and parks a handler over
+        # this healthy connection.
+        with self._reconnection_lock:
+            if self._reconnection_handler:
+                self._reconnection_handler.cancel()
+                self._reconnection_handler = None
+
+            # A connection is in place, so an error on it must be able to queue
+            # a fresh attempt rather than be collapsed into the one ending here.
+            self._reconnect_pending = False
+            self._reconnect_requested = False
+
+            with self._lock:
+                if self._is_shutdown:
+                    # shutdown() won the race, so nothing would ever use this
+                    # connection or close it on our behalf.
+                    old, orphan = None, conn
+                else:
+                    old, orphan = self._connection, None
+                    self._connection = conn
+                    self._reconnection_exhausted_connection = None
+                    self._connection_generation += 1
 
         if old:
             log.debug("[control connection] Closing old connection %r, replacing with %r", old, conn)
             old.close()
 
+        if orphan:
+            log.debug("[control connection] Control connection is shut down, closing new connection %r", orphan)
+            orphan.close()
+
     def _attach_application_session(self, keyspace, session):
         """Bind application use of the control connection to ``keyspace``.
 
@@ -4384,44 +4581,177 @@ def reconnect(self):
         if self._is_shutdown:
             return
 
-        self._submit(self._reconnect)
+        # Collapse attempts that are queued but have not started yet. Without
+        # this, a burst of errors queues one _reconnect() each, and every one
+        # of them past the first cancels the reconnection handler the previous
+        # one installed and restarts its backoff schedule.
+        #
+        # An in-flight reconnection handler is already retrying on its own
+        # schedule, and _reconnect() would cancel it and restart that schedule
+        # from its initial delay. The check lives here rather than in the
+        # callers so that it covers every entry point: return_connection() is
+        # driven by the heartbeat and fires once per idle_heartbeat_interval
+        # for as long as the control connection stays defunct, which would
+        # otherwise reset the backoff on every pass and stop it ever growing.
+        # The lock is an RLock, so callers already holding it re-enter safely.
+        with self._reconnection_lock:
+            if self._reconnection_handler is not None:
+                handler = self._reconnection_handler
+                if handler._is_running:
+                    handler._reconnect_requested = True
+
+                # A handler can have been started proactively while the
+                # current connection was healthy. If that connection fails
+                # during the handler's backoff, the remaining proactive retry
+                # still goes first, but exhaustion must hand this independent
+                # failure a fresh schedule. Record it once so recurring
+                # heartbeats for an already-failed connection do not extend a
+                # finite schedule indefinitely.
+                connection = self._connection
+                if (handler._failed_connection is None and
+                        connection is not None and
+                        (connection.is_defunct or connection.is_closed)):
+                    handler._failed_connection = connection
+                    handler._new_failure_detected = True
+                log.debug("[control connection] Reconnection already in progress, "
+                          "not starting another one")
+                return
+            if self._reconnect_pending:
+                log.debug("[control connection] A reconnection attempt is "
+                          "already queued")
+                # Do not discard this trigger. The pending attempt may fail
+                # before it installs a handler that owns future retries.
+                self._reconnect_requested = True
+                return
+            pending_seq = self._raise_reconnect_pending()
+
+        submitted = None
+        try:
+            submitted = self._submit(self._reconnect)
+        finally:
+            if submitted is None:
+                # Nothing was queued, so nothing will clear the flag. This has
+                # to hold even when the submission raised, or no further
+                # reconnection would ever be attempted.
+                self._clear_reconnect_pending(pending_seq)
+
+    def _raise_reconnect_pending(self):
+        """
+        Mark a reconnection attempt as pending and return a token identifying
+        it. Only the holder of the newest token may clear the flag again.
+        """
+        with self._reconnection_lock:
+            self._reconnect_pending = True
+            self._reconnect_pending_seq += 1
+            return self._reconnect_pending_seq
+
+    def _clear_reconnect_pending(self, pending_seq):
+        """
+        Clear the pending flag, unless a newer attempt has claimed it since
+        `pending_seq` was handed out -- that attempt is the one the flag now
+        stands for, and dropping it would let a burst of errors queue several
+        concurrent reconnects.
+        """
+        with self._reconnection_lock:
+            if self._reconnect_pending_seq == pending_seq:
+                self._reconnect_pending = False
+                self._reconnect_requested = False
+
+    def _finish_reconnect(self, pending_seq):
+        """
+        Finish an active attempt, preserving any trigger it collapsed unless
+        a connection or reconnection handler now owns future work.
+        """
+        follow_up_seq = None
+        with self._reconnection_lock:
+            if self._reconnect_pending_seq != pending_seq:
+                return
+
+            if (self._reconnection_handler is not None or
+                    not self._reconnect_requested):
+                self._reconnect_pending = False
+                self._reconnect_requested = False
+                return
+
+            # Keep the pending flag raised while handing the retained trigger
+            # to the executor, so another burst still collapses into this one.
+            self._reconnect_requested = False
+            self._reconnect_pending_seq += 1
+            follow_up_seq = self._reconnect_pending_seq
+
+        submitted = None
+        try:
+            submitted = self._submit(self._reconnect)
+        finally:
+            if submitted is None:
+                self._clear_reconnect_pending(follow_up_seq)
 
     def _reconnect(self):
+        # _reconnect_pending stays set for as long as this attempt runs.
+        # _reconnect_internal() walks the whole query plan twice with a DNS
+        # re-resolution in between, which routinely outlasts
+        # idle_heartbeat_interval when the cluster is unreachable; clearing the
+        # flag here would let every heartbeat in that window queue another
+        # attempt, each one cancelling the handler the previous one installed
+        # and restarting its backoff from the initial delay.
+        pending_seq = self._raise_reconnect_pending()
+
+        with self._lock:
+            generation = self._connection_generation
+
         log.debug("[control connection] Attempting to reconnect")
         try:
             self._set_new_connection(self._reconnect_internal())
-        except NoHostAvailable:
+        except (NoHostAvailable, UnresolvableContactPoints):
             # make a retry schedule (which includes backoff)
             schedule = self._cluster.reconnection_policy.new_schedule()
 
             with self._reconnection_lock:
+                with self._lock:
+                    if self._connection_generation != generation:
+                        # An attempt that overlapped this one installed a
+                        # connection while we were failing. Parking a handler
+                        # now would block every later reconnect() for the whole
+                        # backoff and then replace a healthy connection.
+                        log.debug("[control connection] Reconnect failed but "
+                                  "another attempt succeeded, not scheduling "
+                                  "retries")
+                        return
 
                 # cancel existing reconnection attempts
                 if self._reconnection_handler:
                     self._reconnection_handler.cancel()
 
                 # when a connection is successfully made, _set_new_connection
-                # will be called with the new connection and then our
-                # _reconnection_handler will be cleared out
-                self._reconnection_handler = _ControlReconnectionHandler(
-                    self, self._cluster.scheduler, schedule,
-                    self._get_and_set_reconnection_handler,
-                    new_handler=None)
-                self._reconnection_handler.start()
+                # will be called with the new connection and will clear out
+                # our _reconnection_handler
+                handler = _ControlReconnectionHandler(
+                    self, self._cluster.scheduler, schedule)
+                self._reconnection_handler = handler
+                try:
+                    handler.start()
+                except StopIteration:
+                    # An empty schedule means that the policy permits no
+                    # retries. Treat it like a finite schedule exhausted by a
+                    # failed handler run so heartbeats for the same defunct
+                    # connection do not start it over indefinitely.
+                    handler._release(exhausted=True)
+                except Exception:
+                    # A handler that never started would sit in the slot
+                    # forever, and reconnect() would take it for one still
+                    # retrying.
+                    if self._reconnection_handler is handler:
+                        self._reconnection_handler = None
+                    raise
         except Exception:
             log.debug("[control connection] error reconnecting", exc_info=True)
             raise
-
-    def _get_and_set_reconnection_handler(self, new_handler):
-        """
-        Called by the _ControlReconnectionHandler when a new connection
-        is successfully created.  Clears out the _reconnection_handler on
-        this ControlConnection.
-        """
-        with self._reconnection_lock:
-            old = self._reconnection_handler
-            self._reconnection_handler = new_handler
-            return old
+        finally:
+            # The attempt is over either way: a connection is installed (the
+            # flag was already cleared as it went in), a handler is parked and
+            # collapses later attempts on its own, or a retained trigger queues
+            # one follow-up after a failure that left no durable retry work.
+            self._finish_reconnect(pending_seq)
 
     def _submit(self, *args, **kwargs):
         try:
@@ -4978,51 +5308,19 @@ def _signal_error(self):
             if self._is_shutdown:
                 return
 
-            # try just signaling the cluster, as this will trigger a reconnect
-            # as part of marking the host down
+            # If DOWN handling is dispatched, its control connection callback
+            # will reconnect. Otherwise reconnect directly.
             if self._connection and self._connection.is_defunct:
                 connection = self._connection
                 host = self._get_host_for_connection(connection)
-                # host may be None if it's already been removed, but that indicates
-                # that errors have already been reported, so we're fine
-                if host:
-                    original_endpoint = getattr(
-                        connection, 'original_endpoint', None)
-                    unix_backed = (
-                        isinstance(host.endpoint, UnixSocketEndPoint) or
-                        isinstance(connection.endpoint, UnixSocketEndPoint) or
-                        isinstance(original_endpoint, UnixSocketEndPoint))
-                    route_mismatch = connection.endpoint != host.endpoint
-                    # Keep ordinary endpoint-equal TCP connections on the
-                    # legacy signal-only path. General suppressed-DOWN recovery
-                    # and its reconnection cadence are outside this change.
-                    if not unix_backed and not route_mismatch:
-                        self._cluster.signal_connection_failure(
-                            host, connection.last_error,
-                            is_host_addition=False)
-                        return
-
-                    # A newly resolvable Unix Host or alternate connection
-                    # route still needs the direct reconnect fallback when
-                    # host-state handling suppresses its DOWN notification. A
-                    # fresh DOWN transition guarantees that on_down() will
-                    # enqueue the reconnect instead.
-                    with host.lock:
-                        host_was_up = host.is_up is True
-                        host_was_reconnecting = (
-                            host.is_currently_reconnecting())
-                        self._cluster.signal_connection_failure(
-                            host, connection.last_error,
-                            is_host_addition=False)
-                        down_notification_queued = (
-                            host_was_up and not host_was_reconnecting and
-                            host.is_up is False)
-
-                    if down_notification_queued:
-                        return
+                if host and self._cluster.signal_connection_failure(
+                        host, connection.last_error,
+                        is_host_addition=False):
+                    return
 
-        # if the connection is not defunct or the host already left, reconnect
-        # manually
+        # If the connection is not defunct, the host is unresolved, or DOWN
+        # handling was suppressed, reconnect manually. reconnect() leaves an
+        # in-flight reconnection handler alone on its own schedule.
         self.reconnect()
 
     def on_up(self, host):
@@ -5031,12 +5329,13 @@ def on_up(self, host):
     def on_down(self, host):
 
         conn = self._connection
-        if self._connection_matches_host(conn, host) and \
-                self._reconnection_handler is None:
-            log.debug("[control connection] Control connection host (%s) is "
-                      "considered down, starting reconnection", host)
-            # this will result in a task being submitted to the executor to reconnect
-            self.reconnect()
+        if not self._connection_matches_host(conn, host):
+            return
+
+        log.debug("[control connection] Control connection host (%s) is "
+                  "considered down, starting reconnection", host)
+        # this will result in a task being submitted to the executor to reconnect
+        self.reconnect()
 
     def on_add(self, host, refresh_nodes=True):
         if refresh_nodes:
@@ -5057,6 +5356,11 @@ def get_connections(self):
 
     def return_connection(self, connection):
         if connection is self._connection and (connection.is_defunct or connection.is_closed):
+            with self._reconnection_lock:
+                if connection is self._reconnection_exhausted_connection:
+                    log.debug("[control connection] Reconnection schedule is "
+                              "exhausted for the defunct connection")
+                    return
             self.reconnect()
 
 
diff --git a/cassandra/pool.py b/cassandra/pool.py
index 8cc22018d2..d4a11009c6 100644
--- a/cassandra/pool.py
+++ b/cassandra/pool.py
@@ -267,6 +267,11 @@ class _ReconnectionHandler(object):
 
     _cancelled = False
 
+    # Whether on_reconnection() keeps the connection it is handed. A handler
+    # that only uses it to probe the host leaves this False and run() closes
+    # the connection for it.
+    _keeps_connection = False
+
     def __init__(self, scheduler, schedule, callback, *callback_args, **callback_kwargs):
         self.scheduler = scheduler
         self.schedule = schedule
@@ -287,6 +292,7 @@ def run(self):
             return
 
         conn = None
+        handed_off = False
         try:
             conn = self.try_reconnect()
         except Exception as exc:
@@ -306,10 +312,16 @@ def run(self):
                     self.scheduler.schedule(next_delay, self.run)
         else:
             if not self._cancelled:
+                # Mark the handoff before it happens: on_reconnection() adopts
+                # the connection and may then raise (installing the new control
+                # connection closes the old one, which runs user callbacks). If
+                # the flag were set afterwards, that raise would leave us
+                # closing a connection the subclass is already using.
+                handed_off = self._keeps_connection
                 self.on_reconnection(conn)
                 self.callback(*(self.callback_args), **(self.callback_kwargs))
         finally:
-            if conn:
+            if conn and not handed_off:
                 conn.close()
 
     def cancel(self):
diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py
index 4792083817..9f6dfb5403 100644
--- a/tests/unit/test_cluster.py
+++ b/tests/unit/test_cluster.py
@@ -1204,6 +1204,265 @@ def test_set_keyspace_for_all_pools_reports_all_errors(self, *_):
         callback.assert_called_once()
         assert callback.call_args.args[0] == {'host1': [keyspace_error]}
 
+
+class ClusterDownHandlingTest(unittest.TestCase):
+
+    def setUp(self):
+        self.cluster = Cluster(contact_points=[])
+        self.addCleanup(self.cluster.shutdown)
+        self.cluster.profile_manager.distance = Mock(
+            return_value=HostDistance.LOCAL)
+        # The real method reports whether the executor accepted the work.
+        self.cluster.on_down_potentially_blocking = Mock(return_value=True)
+        self.cluster._restart_reconnector = Mock(return_value=None)
+        self.host = Host(
+            "127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4())
+        self.host.set_up()
+
+    def test_down_handling_reaches_control_connection_despite_a_raising_policy(self):
+        # signal_connection_failure() reports DOWN handling as dispatched, and
+        # the control connection relies on that to mean its own reconnection
+        # will be started. A user policy raising must not be what prevents it.
+        del self.cluster.on_down_potentially_blocking  # use the real method
+        self.cluster.profile_manager.on_down = Mock(
+            side_effect=RuntimeError('load balancing policy failed'))
+        self.cluster.control_connection = Mock()
+        self.cluster._start_reconnector = Mock()
+        # The method runs in the executor; call the body directly so nothing
+        # is swallowed by the future.
+        body = Cluster.on_down_potentially_blocking.__wrapped__
+
+        body(self.cluster, self.host, is_host_addition=False,
+             down_event_generation=self.host._down_event_generation)
+
+        self.cluster.control_connection.on_down.assert_called_once_with(
+            self.host)
+        self.cluster._start_reconnector.assert_called_once()
+
+    def test_down_handling_continues_when_the_control_connection_raises(self):
+        # The control connection goes first, so a failure there -- the executor
+        # rejecting its reconnect submission, say -- must not skip the pool and
+        # listener work or the host reconnector that follows it.
+        del self.cluster.on_down_potentially_blocking  # use the real method
+        self.cluster.control_connection = Mock()
+        self.cluster.control_connection.on_down.side_effect = RuntimeError(
+            'cannot schedule new futures after shutdown')
+        self.cluster._start_reconnector = Mock()
+        self.cluster.profile_manager.on_down = Mock()
+        body = Cluster.on_down_potentially_blocking.__wrapped__
+
+        body(self.cluster, self.host, is_host_addition=False,
+             down_event_generation=self.host._down_event_generation)
+
+        self.cluster.profile_manager.on_down.assert_called_once_with(self.host)
+        self.cluster._start_reconnector.assert_called_once_with(
+            self.host, False)
+
+    def test_down_handling_continues_when_a_session_raises(self):
+        del self.cluster.on_down_potentially_blocking  # use the real method
+        failing_session = Mock()
+        failing_session.on_down.side_effect = RuntimeError(
+            'session down handling failed')
+        following_session = Mock()
+        self.cluster.sessions = [failing_session, following_session]
+        self.cluster._start_reconnector = Mock()
+        body = Cluster.on_down_potentially_blocking.__wrapped__
+
+        body(self.cluster, self.host, is_host_addition=False,
+             down_event_generation=self.host._down_event_generation)
+
+        following_session.on_down.assert_called_once_with(self.host)
+        self.cluster._start_reconnector.assert_called_once_with(
+            self.host, False)
+
+    def test_down_handling_continues_when_a_listener_raises(self):
+        del self.cluster.on_down_potentially_blocking  # use the real method
+        failing_listener = Mock()
+        failing_listener.on_down.side_effect = RuntimeError(
+            'listener down handling failed')
+        following_listener = Mock()
+        self.cluster._listeners = {failing_listener, following_listener}
+        self.cluster._start_reconnector = Mock()
+        body = Cluster.on_down_potentially_blocking.__wrapped__
+
+        body(self.cluster, self.host, is_host_addition=False,
+             down_event_generation=self.host._down_event_generation)
+
+        following_listener.on_down.assert_called_once_with(self.host)
+        self.cluster._start_reconnector.assert_called_once_with(
+            self.host, False)
+
+    def test_signal_connection_failure_rejected_by_conviction_policy(self):
+        error = ConnectionException("connection failed")
+        self.host.signal_connection_failure = Mock(return_value=False)
+        self.cluster.on_down = Mock(return_value=True)
+
+        assert not self.cluster.signal_connection_failure(
+            self.host, error, is_host_addition=False)
+
+        self.host.signal_connection_failure.assert_called_once_with(error)
+        self.cluster.on_down.assert_not_called()
+
+    def test_signal_connection_failure_returns_down_handling_result(self):
+        error = ConnectionException("connection failed")
+        self.host.signal_connection_failure = Mock(return_value=True)
+        self.cluster.on_down = Mock(return_value=False)
+
+        assert not self.cluster.signal_connection_failure(
+            self.host, error, is_host_addition=True,
+            expect_host_to_be_down=True)
+
+        self.cluster.on_down.assert_called_once_with(
+            self.host, True, True)
+
+    def test_on_down_dispatches_normal_up_to_down_transition(self):
+        assert self.cluster.on_down(self.host, is_host_addition=False)
+
+        assert self.host.is_up is False
+        self.cluster.on_down_potentially_blocking.assert_called_once_with(
+            self.host, False, self.host._down_event_generation)
+
+    def test_on_down_discounted_when_any_session_has_an_open_pool(self):
+        missing_pool = Mock()
+        missing_pool.get_pool_state.return_value = {}
+        closed_pool = Mock()
+        closed_pool.get_pool_state.return_value = {
+            self.host: {'open_count': 0}}
+        open_pool = Mock()
+        open_pool.get_pool_state.return_value = {
+            self.host: {'open_count': 1}}
+        self.cluster.sessions = [missing_pool, closed_pool, open_pool]
+
+        assert not self.cluster.on_down(self.host, is_host_addition=False)
+
+        assert self.host.is_up is True
+        self.cluster.on_down_potentially_blocking.assert_not_called()
+
+    def test_on_down_not_discounted_when_all_pools_are_missing_or_closed(self):
+        missing_pool = Mock()
+        missing_pool.get_pool_state.return_value = {}
+        closed_pool = Mock()
+        closed_pool.get_pool_state.return_value = {
+            self.host: {'open_count': 0}}
+        self.cluster.sessions = [missing_pool, closed_pool]
+
+        assert self.cluster.on_down(self.host, is_host_addition=False)
+
+        self.cluster.on_down_potentially_blocking.assert_called_once_with(
+            self.host, False, self.host._down_event_generation)
+
+    def test_on_down_open_pool_discount_can_be_disabled(self):
+        open_pool = Mock()
+        open_pool.get_pool_state.return_value = {
+            self.host: {'open_count': 1}}
+        self.cluster.sessions = [open_pool]
+        self.cluster._discount_down_events = False
+
+        assert self.cluster.on_down(self.host, is_host_addition=False)
+
+        self.cluster.on_down_potentially_blocking.assert_called_once_with(
+            self.host, False, self.host._down_event_generation)
+
+    def test_on_down_does_not_discount_ignored_host(self):
+        open_pool = Mock()
+        open_pool.get_pool_state.return_value = {
+            self.host: {'open_count': 1}}
+        self.cluster.sessions = [open_pool]
+        self.cluster.profile_manager.distance.return_value = \
+            HostDistance.IGNORED
+
+        assert self.cluster.on_down(self.host, is_host_addition=False)
+
+        self.cluster.on_down_potentially_blocking.assert_called_once_with(
+            self.host, False, self.host._down_event_generation)
+
+    def test_on_down_skipped_during_shutdown(self):
+        self.cluster.is_shutdown = True
+
+        try:
+            assert not self.cluster.on_down(
+                self.host, is_host_addition=False)
+        finally:
+            # Keep the real shutdown cleanup effective for this test cluster.
+            self.cluster.is_shutdown = False
+
+        assert self.host.is_up is True
+        self.cluster.on_down_potentially_blocking.assert_not_called()
+
+    def test_on_down_skipped_when_pool_creation_is_disabled(self):
+        self.cluster.allow_control_connection_query_fallback = \
+            ControlConnectionQueryFallback.SkipPoolCreation
+
+        assert not self.cluster.on_down(self.host, is_host_addition=False)
+
+        assert self.host.is_up is True
+        self.cluster.on_down_potentially_blocking.assert_not_called()
+
+    def test_on_down_restarts_already_down_or_uninitialized_host(self):
+        for initial_state in (False, None):
+            with self.subTest(initial_state=initial_state):
+                self.host.is_up = initial_state
+                self.host._currently_handling_node_down = False
+                self.cluster.on_down_potentially_blocking.reset_mock()
+                self.cluster._restart_reconnector.reset_mock()
+
+                assert not self.cluster.on_down(
+                    self.host, is_host_addition=False)
+
+                assert self.host.is_up is False
+                self.cluster.on_down_potentially_blocking.assert_not_called()
+                self.cluster._restart_reconnector.assert_called_once_with(
+                    self.host, False, self.host._down_event_generation)
+
+    def test_on_down_expected_down_host_dispatches_recovery(self):
+        for initial_state in (False, None):
+            with self.subTest(initial_state=initial_state):
+                self.host.is_up = initial_state
+                self.host._currently_handling_node_down = False
+                self.cluster.on_down_potentially_blocking.reset_mock()
+
+                assert self.cluster.on_down(
+                    self.host, is_host_addition=True,
+                    expect_host_to_be_down=True)
+
+                self.cluster.on_down_potentially_blocking \
+                    .assert_called_once_with(
+                        self.host, True, self.host._down_event_generation)
+
+    def test_on_down_dispatches_transition_with_existing_reconnector(self):
+        self.host.get_and_set_reconnection_handler(Mock())
+
+        assert self.cluster.on_down(self.host, is_host_addition=False)
+
+        assert self.host.is_up is False
+        self.cluster.on_down_potentially_blocking.assert_called_once_with(
+            self.host, False, self.host._down_event_generation)
+
+    def test_on_down_reports_undispatched_when_the_executor_rejects_it(self):
+        # Restore the real method so the executor submission is exercised.
+        del self.cluster.on_down_potentially_blocking
+        real_executor = self.cluster.executor
+        self.addCleanup(setattr, self.cluster, 'executor', real_executor)
+        self.cluster.executor = Mock()
+        self.cluster.executor.submit.side_effect = RuntimeError(
+            "cannot schedule new futures")
+
+        assert not self.cluster.on_down(self.host, is_host_addition=False)
+
+        assert self.host.is_up is False
+        self.cluster.executor.submit.assert_called_once()
+
+    def test_on_down_reports_dispatched_when_the_executor_accepts_it(self):
+        del self.cluster.on_down_potentially_blocking
+        real_executor = self.cluster.executor
+        self.addCleanup(setattr, self.cluster, 'executor', real_executor)
+        self.cluster.executor = Mock()
+
+        assert self.cluster.on_down(self.host, is_host_addition=False)
+
+        self.cluster.executor.submit.assert_called_once()
+
+
 class ProtocolVersionTests(unittest.TestCase):
 
     def test_protocol_downgrade_test(self):
diff --git a/tests/unit/test_control_connection.py b/tests/unit/test_control_connection.py
index 13a68c7e55..190aee1b18 100644
--- a/tests/unit/test_control_connection.py
+++ b/tests/unit/test_control_connection.py
@@ -12,21 +12,30 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import gc
 import unittest
+import weakref
 
 from concurrent.futures import ThreadPoolExecutor
+from threading import Event, Thread
 from unittest.mock import Mock, ANY, call, patch
 
-from cassandra import OperationTimedOut, SchemaTargetType, SchemaChangeType
+from cassandra import (AuthenticationFailed, OperationTimedOut,
+                       SchemaTargetType, SchemaChangeType,
+                       UnresolvableContactPoints)
 from cassandra.protocol import ResultMessage, RESULT_KIND_ROWS
 from cassandra.cluster import (Cluster, ControlConnection, _Scheduler,
                                ProfileManager, EXEC_PROFILE_DEFAULT,
-                               ExecutionProfile)
-from cassandra.pool import Host
+                               ExecutionProfile,
+                               ControlConnectionQueryFallback,
+                               NoHostAvailable, _ControlReconnectionHandler)
+from cassandra.pool import Host, _ReconnectionHandler
 from cassandra.connection import (ConnectionException, EndPoint, DefaultEndPoint,
                                   DefaultEndPointFactory, UnixSocketEndPoint)
-from cassandra.policies import (SimpleConvictionPolicy, RoundRobinPolicy,
-                                ConstantReconnectionPolicy, IdentityTranslator)
+from cassandra.policies import (HostDistance, SimpleConvictionPolicy,
+                                RoundRobinPolicy, ConstantReconnectionPolicy,
+                                ExponentialReconnectionPolicy,
+                                IdentityTranslator)
 
 PEER_IP = "foobar"
 
@@ -149,6 +158,7 @@ def _node_meta_results(local_results, peer_results):
 class MockConnection(object):
 
     is_defunct = False
+    is_closed = False
 
     def __init__(self):
         self.endpoint = DefaultEndPoint("192.168.1.0")
@@ -230,6 +240,29 @@ def _refresh_control_connection_over_network(self):
         self.connection.original_endpoint = self.connection.endpoint
         self.control_connection.refresh_node_list_and_token_map()
 
+    def _use_cluster_down_handling(
+            self, sessions=(),
+            fallback=ControlConnectionQueryFallback.Disabled):
+        self.cluster.sessions = list(sessions)
+        self.cluster._discount_down_events = True
+        self.cluster.allow_control_connection_query_fallback = fallback
+        self.cluster.profile_manager = Mock()
+        self.cluster.profile_manager.distance.return_value = \
+            HostDistance.LOCAL
+        # The real method reports whether the executor accepted the work.
+        self.cluster.on_down_potentially_blocking = Mock(return_value=True)
+        self.cluster._restart_reconnector = Mock(return_value=True)
+        self.cluster.on_down = Cluster.on_down.__get__(self.cluster)
+        self.cluster.signal_connection_failure = \
+            Cluster.signal_connection_failure.__get__(self.cluster)
+
+    def _discount_down_for(self, host):
+        """Model a session pool that keeps ``host`` up despite a conviction."""
+        session = Mock()
+        session.get_pool_state.return_value = {host: {'open_count': 1}}
+        self._use_cluster_down_handling([session])
+        return session
+
     def test_wait_for_schema_agreement(self):
         """
         Basic test with all schema versions agreeing
@@ -469,6 +502,804 @@ def test_schema_query_uses_shard_aware_connection_original_endpoint(self):
         assert query == self.control_connection._SELECT_SCHEMA_PEERS_TEMPLATE \
             .format(nt_col_name='rpc_address')
 
+    def test_defunct_tcp_control_reconnects_when_open_pool_discounts_down(self):
+        host = self.cluster.metadata.get_host_by_host_id('uuid1')
+        host.set_up()
+        session = Mock()
+        session.get_pool_state.return_value = {
+            host: {'open_count': 1}}
+        self._use_cluster_down_handling([session])
+        self.connection.is_defunct = True
+        self.connection.last_error = ConnectionException(
+            'control connection failed')
+        self.cluster.executor.reset_mock()
+
+        self.control_connection._signal_error()
+
+        assert host.is_up is True
+        self.cluster.on_down_potentially_blocking.assert_not_called()
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_defunct_control_reconnects_when_conviction_is_rejected(self):
+        host = self.cluster.metadata.get_host_by_host_id('uuid1')
+        host.set_up()
+        self._use_cluster_down_handling()
+        self.connection.is_defunct = True
+        self.connection.last_error = OperationTimedOut()
+        self.cluster.executor.reset_mock()
+
+        self.control_connection._signal_error()
+
+        assert host.is_up is True
+        self.cluster.on_down_potentially_blocking.assert_not_called()
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_defunct_control_reconnects_when_host_is_already_down(self):
+        host = self.cluster.metadata.get_host_by_host_id('uuid1')
+        host.set_down()
+        self._use_cluster_down_handling()
+        self.connection.is_defunct = True
+        self.connection.last_error = ConnectionException(
+            'control connection failed')
+        self.cluster.executor.reset_mock()
+
+        self.control_connection._signal_error()
+
+        self.cluster.on_down_potentially_blocking.assert_not_called()
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_defunct_control_reconnects_when_host_reconnector_is_active(self):
+        host = self.cluster.metadata.get_host_by_host_id('uuid1')
+        host.set_down()
+        host.get_and_set_reconnection_handler(Mock())
+        self._use_cluster_down_handling()
+        self.connection.is_defunct = True
+        self.connection.last_error = ConnectionException(
+            'control connection failed')
+        self.cluster.executor.reset_mock()
+
+        self.control_connection._signal_error()
+
+        assert host.is_up is False
+        self.cluster.on_down_potentially_blocking.assert_not_called()
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_defunct_control_reconnects_when_pool_creation_is_disabled(self):
+        host = self.cluster.metadata.get_host_by_host_id('uuid1')
+        host.set_up()
+        self._use_cluster_down_handling(
+            fallback=ControlConnectionQueryFallback.SkipPoolCreation)
+        self.connection.is_defunct = True
+        self.connection.last_error = ConnectionException(
+            'control connection failed')
+        self.cluster.executor.reset_mock()
+
+        self.control_connection._signal_error()
+
+        assert host.is_up is True
+        self.cluster.on_down_potentially_blocking.assert_not_called()
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_defunct_control_waits_for_dispatched_down_callback(self):
+        host = self.cluster.metadata.get_host_by_host_id('uuid1')
+        host.set_up()
+        self._use_cluster_down_handling()
+        self.connection.is_defunct = True
+        self.connection.last_error = ConnectionException(
+            'control connection failed')
+        self.cluster.executor.reset_mock()
+
+        self.control_connection._signal_error()
+
+        assert host.is_up is False
+        self.cluster.on_down_potentially_blocking.assert_called_once_with(
+            host, False, host._down_event_generation)
+        self.cluster.executor.submit.assert_not_called()
+
+        self.control_connection.on_down(host)
+
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_signal_error_reconnects_non_defunct_connection(self):
+        self.connection.is_defunct = False
+        self.cluster.executor.reset_mock()
+
+        self.control_connection._signal_error()
+
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_signal_error_reconnects_when_host_is_unresolved(self):
+        self._forget_local_host()
+        self.connection.is_defunct = True
+        self.connection.last_error = ConnectionException(
+            'control connection failed')
+        self.cluster.executor.reset_mock()
+
+        self.control_connection._signal_error()
+
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_signal_error_does_nothing_after_control_connection_shutdown(self):
+        self.control_connection._is_shutdown = True
+        self.connection.is_defunct = True
+        self.cluster.executor.reset_mock()
+
+        self.control_connection._signal_error()
+
+        self.cluster.executor.submit.assert_not_called()
+
+    def test_signal_error_leaves_an_in_flight_reconnection_alone(self):
+        # _reconnect() cancels the handler and restarts its schedule from the
+        # initial delay, so repeated errors must not keep resetting the backoff.
+        host = self.cluster.metadata.get_host_by_host_id('uuid1')
+        host.set_down()
+        self._use_cluster_down_handling()
+        self.control_connection._reconnection_handler = Mock()
+        self.connection.is_defunct = True
+        self.connection.last_error = ConnectionException(
+            'control connection failed')
+        self.cluster.executor.reset_mock()
+
+        self.control_connection._signal_error()
+
+        self.cluster.executor.submit.assert_not_called()
+
+    def _make_reconnection_handler(self):
+        handler = _ControlReconnectionHandler(
+            self.control_connection, self.cluster.scheduler, iter([1.0]))
+        self.control_connection._reconnection_handler = handler
+        return handler
+
+    def test_reconnection_handler_releases_its_slot_when_it_gives_up(self):
+        handler = self._make_reconnection_handler()
+
+        handler.on_exception(ConnectionException('refused'), None)
+
+        assert self.control_connection._reconnection_handler is None
+
+    def test_reconnection_handler_keeps_its_slot_while_it_retries(self):
+        for exc in (ConnectionException('refused'),
+                    AuthenticationFailed('bad password')):
+            with self.subTest(exc=exc):
+                handler = self._make_reconnection_handler()
+
+                assert handler.on_exception(exc, 1.0)
+
+                assert self.control_connection._reconnection_handler is handler
+
+    def test_reconnection_handler_never_releases_a_replacement(self):
+        handler = self._make_reconnection_handler()
+        replacement = self._make_reconnection_handler()
+
+        handler.on_exception(ConnectionException('refused'), None)
+
+        assert self.control_connection._reconnection_handler is replacement
+
+    def test_reconnection_handler_run_leaves_a_replacement_in_the_slot(self):
+        # A finishing handler must not clear a replacement that another thread
+        # parked in the slot while it was handing its connection over.
+        # _set_new_connection() has already released this handler by then, so
+        # clearing again can only evict someone else -- and it would do so
+        # without cancelling them, leaving a handler that keeps retrying where
+        # reconnect() can no longer see it.
+        handler = self._make_reconnection_handler()
+        self.control_connection._connection = None
+        conn = Mock()
+        replacement = []
+
+        def install_and_lose_the_race(new_conn):
+            # What _set_new_connection() really does -- release the slot and
+            # adopt the connection -- plus another thread winning the race to
+            # park a fresh handler in the slot it just emptied.
+            self.control_connection._reconnection_handler = None
+            self.control_connection._connection = new_conn
+            replacement.append(self._make_reconnection_handler())
+
+        with patch.object(handler, 'try_reconnect', return_value=conn), \
+                patch.object(self.control_connection, '_set_new_connection',
+                             side_effect=install_and_lose_the_race):
+            handler.run()
+
+        assert self.control_connection._reconnection_handler is replacement[0]
+
+    def test_reconnection_handler_keeps_the_connection_it_installs(self):
+        # run() closes the connection it opened, which is right for the host
+        # handler that only probes with it, but this one hands it to the
+        # control connection.
+        handler = self._make_reconnection_handler()
+        self.control_connection._connection = None
+        conn = Mock()
+
+        with patch.object(handler, 'try_reconnect', return_value=conn):
+            handler.run()
+
+        assert self.control_connection._connection is conn
+        conn.close.assert_not_called()
+
+    def test_reconnection_handler_keeps_the_connection_a_failed_install_took(self):
+        # Installing the new control connection closes the old one, which runs
+        # user callbacks, and one of those may raise. The connection is already
+        # adopted by then, so it must not be closed on the way out.
+        handler = self._make_reconnection_handler()
+        self.control_connection._connection = None
+        conn = Mock()
+
+        with patch.object(handler, 'try_reconnect', return_value=conn), \
+                patch.object(self.control_connection, '_set_new_connection',
+                             side_effect=RuntimeError('listener failed')):
+            with self.assertRaises(RuntimeError):
+                handler.run()
+
+        conn.close.assert_not_called()
+
+    def test_reconnection_handler_closes_a_connection_it_only_probes_with(self):
+        # The plain handler, like the host one, only uses the connection to
+        # prove the host is reachable, so run() still closes it for it.
+        handler = _ReconnectionHandler(self.cluster.scheduler, iter([1.0]),
+                                       Mock())
+        conn = Mock()
+
+        with patch.object(handler, 'try_reconnect', return_value=conn):
+            handler.run()
+
+        conn.close.assert_called_once_with()
+
+    def test_reconnecting_successfully_releases_a_parked_handler(self):
+        # A handler installed by an earlier failure is still backing off when
+        # an unrelated reconnect succeeds. Left in the slot it would look like
+        # a reconnection in progress to every later error.
+        handler = self._make_reconnection_handler()
+        self.control_connection._connection = None
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          return_value=Mock()):
+            self.control_connection._reconnect()
+
+        assert self.control_connection._reconnection_handler is None
+        assert handler._cancelled
+
+    def test_set_new_connection_closes_a_connection_shutdown_beat(self):
+        # shutdown() already closed the control connection, so nothing would
+        # ever use this one or close it on our behalf.
+        self.control_connection._is_shutdown = True
+        conn = Mock()
+
+        self.control_connection._set_new_connection(conn)
+
+        conn.close.assert_called_once_with()
+        assert self.control_connection._connection is self.connection
+
+    def test_signal_error_reconnects_once_a_reconnection_has_given_up(self):
+        host = self.cluster.metadata.get_host_by_host_id('uuid1')
+        host.set_down()
+        self._use_cluster_down_handling()
+        handler = self._make_reconnection_handler()
+        handler.on_exception(ConnectionException('refused'), None)
+        self.connection.is_defunct = True
+        self.connection.last_error = ConnectionException(
+            'control connection failed')
+        self.cluster.executor.reset_mock()
+
+        self.control_connection._signal_error()
+
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_reconnect_collapses_an_attempt_that_has_not_started(self):
+        self.cluster.executor.reset_mock()
+
+        self.control_connection.reconnect()
+        self.control_connection.reconnect()
+
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_reconnect_is_queued_again_once_the_attempt_starts(self):
+        self.cluster.executor.reset_mock()
+        self.control_connection.reconnect()
+        self.control_connection._connection = None
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          return_value=Mock()):
+            self.control_connection._reconnect()
+
+        self.control_connection.reconnect()
+
+        assert self.cluster.executor.submit.call_args_list == [
+            call(self.control_connection._reconnect),
+            call(self.control_connection._reconnect)]
+
+    def test_reconnect_does_not_clear_a_flag_a_newer_attempt_claimed(self):
+        # _set_new_connection() drops the pending flag as the connection goes
+        # in. If that connection errors before the installing attempt returns,
+        # the reconnect() it triggers claims the flag, and the finishing
+        # attempt must not clear it -- a further trigger would then queue a
+        # second attempt that runs alongside the first.
+        self.cluster.executor.reset_mock()
+        self.control_connection._connection = None
+        install = self.control_connection._set_new_connection
+
+        def install_then_lose_the_connection(conn):
+            install(conn)
+            self.control_connection.reconnect()
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          return_value=Mock()), \
+                patch.object(self.control_connection, '_set_new_connection',
+                             side_effect=install_then_lose_the_connection):
+            self.control_connection._reconnect()
+
+        assert self.control_connection._reconnect_pending
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+        self.control_connection.reconnect()
+
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_reconnect_collapses_an_attempt_while_one_is_running(self):
+        # _reconnect_internal() walks the whole query plan and can outlast the
+        # idle heartbeat, which calls reconnect() through return_connection().
+        # Those calls must not queue a second attempt that would cancel the
+        # first one's handler and restart its backoff.
+        def reconnect_while_running():
+            self.cluster.executor.reset_mock()
+            self.control_connection.reconnect()
+            raise NoHostAvailable('no host', {})
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=reconnect_while_running):
+            self.control_connection._reconnect()
+
+        self.cluster.executor.submit.assert_not_called()
+
+    def test_reconnect_handler_owns_trigger_after_dns_failure(self):
+        # A trigger arriving during an attempt must not queue duplicate work
+        # when that attempt leaves a backoff handler owning future retries.
+        def reconnect_while_running():
+            self.cluster.executor.reset_mock()
+            self.control_connection.reconnect()
+            raise UnresolvableContactPoints({})
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=reconnect_while_running):
+            self.control_connection._reconnect()
+
+        self.cluster.executor.submit.assert_not_called()
+        assert self.control_connection._reconnection_handler is not None
+        assert not self.control_connection._reconnect_pending
+
+    def test_reconnect_preserves_trigger_if_new_handler_already_exhausted(self):
+        # A zero-delay finite handler can exhaust on another worker before the
+        # attempt that started it reaches _finish_reconnect(). It no longer
+        # owns retries at that point, so a trigger collapsed in between must
+        # be submitted as follow-up work.
+        self.cluster.reconnection_policy = ConstantReconnectionPolicy(
+            0, max_attempts=1)
+        self.connection.is_defunct = True
+
+        def exhaust_then_trigger(_delay, run):
+            run()
+            self.control_connection.reconnect()
+
+        self.cluster.scheduler.schedule.side_effect = exhaust_then_trigger
+        self.cluster.executor.reset_mock()
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=NoHostAvailable('no host', {})):
+            self.control_connection._reconnect()
+
+        assert self.control_connection._reconnection_handler is None
+        assert self.control_connection._reconnect_pending
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_reconnect_retries_when_contact_points_cannot_resolve(self):
+        # A DNS failure leaves no connection that can trigger a heartbeat, so
+        # it must enter the normal reconnection backoff on its own.
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=UnresolvableContactPoints({})):
+            self.control_connection._reconnect()
+
+        assert self.control_connection._reconnection_handler is not None
+
+    def test_connect_does_not_raise_when_shutdown_beats_it(self):
+        # shutdown() ran while _reconnect_internal() was connecting, so
+        # _set_new_connection() declined to install anything.
+        self.control_connection._connection = None
+        self.cluster.protocol_version = 4
+
+        def shut_down_and_connect():
+            self.control_connection._is_shutdown = True
+            return Mock()
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=shut_down_and_connect):
+            self.control_connection.connect()
+
+        assert self.control_connection._connection is None
+
+    def test_reconnect_does_not_park_a_handler_after_another_attempt_won(self):
+        # Two attempts can overlap: reconnect() only collapses ones that have
+        # not started. If this one fails after the other installed a live
+        # connection, a handler parked here would block every later
+        # reconnect() for the whole backoff and then replace that connection.
+        def lose_the_race():
+            self.control_connection._connection_generation += 1
+            raise NoHostAvailable('no host', {})
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=lose_the_race):
+            self.control_connection._reconnect()
+
+        assert self.control_connection._reconnection_handler is None
+
+    def test_reconnect_parks_a_handler_when_no_attempt_won(self):
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=NoHostAvailable('no host', {})):
+            self.control_connection._reconnect()
+
+        assert self.control_connection._reconnection_handler is not None
+
+    def test_reconnection_handler_closes_the_connection_it_cannot_hand_off(self):
+        # The ControlConnection was collected while the handler was backing
+        # off. run() has already marked the connection as handed off, so
+        # nothing else is left to close it.
+        handler = self._make_reconnection_handler()
+        owner = Mock()
+        handler.control_connection = weakref.proxy(owner)
+        del owner
+        gc.collect()
+        conn = Mock()
+
+        with patch.object(handler, 'try_reconnect', return_value=conn):
+            handler.run()
+
+        conn.close.assert_called_once_with()
+
+    def test_reconnect_defers_to_a_handler_left_by_a_failed_attempt(self):
+        self.cluster.executor.reset_mock()
+        self.control_connection.reconnect()
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=NoHostAvailable('no host', {})):
+            self.control_connection._reconnect()
+
+        handler = self.control_connection._reconnection_handler
+        assert handler is not None
+
+        self.control_connection.reconnect()
+
+        # The handler installed by the failed attempt is retrying on its own
+        # schedule; starting another attempt would cancel it and restart that
+        # schedule from its initial delay.
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+        assert self.control_connection._reconnection_handler is handler
+        assert not handler._cancelled
+
+    def test_reconnect_is_queued_again_after_a_handler_fails_to_start(self):
+        # A handler that never got scheduled retries nothing, so it must not
+        # be left in the slot for reconnect() to defer to forever.
+        self.cluster.scheduler.schedule.side_effect = RuntimeError(
+            'scheduler is shut down')
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=NoHostAvailable('no host', {})):
+            with self.assertRaises(RuntimeError):
+                self.control_connection._reconnect()
+
+        assert self.control_connection._reconnection_handler is None
+
+        self.cluster.executor.reset_mock()
+        self.control_connection.reconnect()
+
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_returning_a_defunct_connection_does_not_restart_the_backoff(self):
+        # The heartbeat hands a defunct control connection back once per
+        # idle_heartbeat_interval for as long as it stays defunct. Each of
+        # those must leave the parked handler's schedule alone.
+        self.cluster.executor.reset_mock()
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=NoHostAvailable('no host', {})):
+            self.control_connection._reconnect()
+
+        handler = self.control_connection._reconnection_handler
+        assert handler is not None
+        self.cluster.executor.reset_mock()
+
+        self.connection.is_defunct = True
+        for _ in range(3):
+            self.control_connection.return_connection(self.connection)
+
+        self.cluster.executor.submit.assert_not_called()
+        assert self.control_connection._reconnection_handler is handler
+        assert not handler._cancelled
+
+    def test_heartbeat_does_not_restart_an_exhausted_finite_schedule(self):
+        self.cluster.reconnection_policy = ConstantReconnectionPolicy(
+            0, max_attempts=1)
+        self.connection.is_defunct = True
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=NoHostAvailable('no host', {})):
+            self.control_connection._reconnect()
+            handler = self.control_connection._reconnection_handler
+            assert handler is not None
+
+            # Run the only scheduled retry. Its failure exhausts the finite
+            # schedule and releases the handler slot.
+            handler.run()
+
+        assert self.control_connection._reconnection_handler is None
+        self.cluster.executor.reset_mock()
+        self.connection.is_defunct = True
+
+        # ConnectionHeartbeat returns this same defunct connection on every
+        # interval; none of those passes may create a new retry schedule.
+        for _ in range(3):
+            self.control_connection.return_connection(self.connection)
+
+        self.cluster.executor.submit.assert_not_called()
+
+    def test_final_handler_attempt_preserves_heartbeat_trigger(self):
+        self.cluster.reconnection_policy = ConstantReconnectionPolicy(
+            0, max_attempts=1)
+        self.connection.is_defunct = True
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=NoHostAvailable('no host', {})):
+            self.control_connection._reconnect()
+
+        handler = self.control_connection._reconnection_handler
+        assert handler is not None
+        self.cluster.executor.reset_mock()
+
+        def fail_after_heartbeat():
+            self.control_connection.return_connection(self.connection)
+            raise NoHostAvailable('no host', {})
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=fail_after_heartbeat):
+            # The only scheduled retry receives a heartbeat trigger while it
+            # is running and then exhausts the finite schedule.
+            handler.run()
+
+        assert self.control_connection._reconnection_handler is None
+        assert self.control_connection._reconnection_exhausted_connection \
+            is None
+        assert self.control_connection._reconnect_pending
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_nonfinal_handler_attempt_consumes_heartbeat_trigger(self):
+        self.cluster.reconnection_policy = ConstantReconnectionPolicy(
+            0, max_attempts=2)
+        self.connection.is_defunct = True
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=NoHostAvailable('no host', {})):
+            self.control_connection._reconnect()
+
+        handler = self.control_connection._reconnection_handler
+        assert handler is not None
+        self.cluster.executor.reset_mock()
+
+        def fail_after_heartbeat():
+            self.control_connection.return_connection(self.connection)
+            raise NoHostAvailable('no host', {})
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=fail_after_heartbeat):
+            # This retry still has another scheduled attempt to own the
+            # heartbeat trigger, so it must not start a new schedule.
+            handler.run()
+
+        assert self.control_connection._reconnection_handler is handler
+        self.cluster.executor.submit.assert_not_called()
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=NoHostAvailable('no host', {})):
+            handler.run()
+
+        assert self.control_connection._reconnection_handler is None
+        assert self.control_connection._reconnection_exhausted_connection \
+            is self.connection
+        self.cluster.executor.submit.assert_not_called()
+
+    def test_older_handler_run_does_not_clear_overlapping_retry_state(self):
+        self.connection.is_defunct = True
+        handler = _ControlReconnectionHandler(
+            self.control_connection, self.cluster.scheduler, iter([0]))
+        self.control_connection._reconnection_handler = handler
+        second_started = Event()
+        finish_second = Event()
+        attempts = [0]
+        retry_thread = []
+
+        def fail_reconnect():
+            attempts[0] += 1
+            if attempts[0] == 1:
+                raise NoHostAvailable('no host', {})
+            second_started.set()
+            finish_second.wait(5)
+            raise NoHostAvailable('no host', {})
+
+        def start_retry(_delay, run):
+            retry_thread.append(Thread(target=run))
+            retry_thread[0].start()
+            assert second_started.wait(5)
+
+        self.cluster.scheduler.schedule.side_effect = start_retry
+        self.cluster.executor.reset_mock()
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=fail_reconnect):
+            handler.run()
+            try:
+                # The zero-delay final retry started before the first run's
+                # finally block. The older run must leave its active state set
+                # so this heartbeat trigger is preserved if the retry fails.
+                assert handler._is_running
+                self.control_connection.return_connection(self.connection)
+            finally:
+                finish_second.set()
+                retry_thread[0].join(5)
+
+        assert not retry_thread[0].is_alive()
+        assert self.control_connection._reconnection_handler is None
+        assert self.control_connection._reconnection_exhausted_connection \
+            is None
+        assert self.control_connection._reconnect_pending
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_heartbeat_does_not_restart_an_empty_reconnection_schedule(self):
+        self.cluster.reconnection_policy = ExponentialReconnectionPolicy(
+            1.0, 2.0, max_attempts=0)
+        self.connection.is_defunct = True
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=NoHostAvailable('no host', {})):
+            self.control_connection._reconnect()
+
+        assert self.control_connection._reconnection_handler is None
+        assert self.control_connection._reconnection_exhausted_connection \
+            is self.connection
+        self.cluster.executor.reset_mock()
+
+        # No retries means that recurring heartbeat returns must not turn the
+        # empty schedule into one fresh immediate attempt per interval.
+        for _ in range(3):
+            self.control_connection.return_connection(self.connection)
+
+        self.cluster.executor.submit.assert_not_called()
+
+    def test_exhausted_replacement_does_not_suppress_a_later_failure(self):
+        self.cluster.reconnection_policy = ConstantReconnectionPolicy(
+            0, max_attempts=1)
+        host = self.cluster.metadata.get_host_by_host_id('uuid1')
+
+        # Removing the connected host starts a proactive replacement while
+        # the existing control connection is still healthy.
+        self.control_connection.on_remove(host)
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=NoHostAvailable('no host', {})):
+            self.control_connection._reconnect()
+            handler = self.control_connection._reconnection_handler
+            assert handler is not None
+            handler.run()
+
+        assert self.control_connection._reconnection_handler is None
+        assert self.control_connection._reconnection_exhausted_connection \
+            is None
+
+        # A later, independent failure of the old connection must get a new
+        # retry schedule rather than being mistaken for the exhausted one.
+        self.cluster.executor.reset_mock()
+        self.connection.is_defunct = True
+        self.control_connection.return_connection(self.connection)
+
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_failure_between_proactive_retries_starts_a_fresh_schedule(self):
+        self.cluster.reconnection_policy = ConstantReconnectionPolicy(
+            0, max_attempts=1)
+        host = self.cluster.metadata.get_host_by_host_id('uuid1')
+
+        # Start a proactive replacement while the existing control connection
+        # is healthy, then park the handler between its scheduled attempts.
+        self.control_connection.on_remove(host)
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=NoHostAvailable('no host', {})):
+            self.control_connection._reconnect()
+
+        handler = self.control_connection._reconnection_handler
+        assert handler is not None
+        assert handler._failed_connection is None
+        self.cluster.executor.reset_mock()
+
+        # With heartbeats disabled, this is the only failure notification the
+        # connection supplies. The proactive handler must retain it until its
+        # own finite schedule exhausts, then start a failure-owned schedule.
+        self.connection.is_defunct = True
+        self.control_connection.return_connection(self.connection)
+        self.cluster.executor.submit.assert_not_called()
+
+        with patch.object(self.control_connection, '_reconnect_internal',
+                          side_effect=NoHostAvailable('no host', {})):
+            handler.run()
+
+        assert self.control_connection._reconnection_handler is None
+        assert self.control_connection._reconnect_pending
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_reconnect_is_queued_again_after_a_rejected_submission(self):
+        self.cluster.executor.reset_mock()
+        self.cluster.is_shutdown = True
+        self.addCleanup(setattr, self.cluster, 'is_shutdown', False)
+
+        self.control_connection.reconnect()
+
+        self.cluster.executor.submit.assert_not_called()
+
+        self.cluster.is_shutdown = False
+        self.control_connection.reconnect()
+
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_reconnect_is_queued_again_after_a_raising_submission(self):
+        self.cluster.executor.reset_mock()
+        self.cluster.executor.submit.side_effect = RuntimeError(
+            'cannot schedule new futures after shutdown')
+
+        with self.assertRaises(RuntimeError):
+            self.control_connection.reconnect()
+
+        self.cluster.executor.submit.side_effect = None
+        self.cluster.executor.reset_mock()
+
+        self.control_connection.reconnect()
+
+        self.cluster.executor.submit.assert_called_once_with(
+            self.control_connection._reconnect)
+
+    def test_defunct_control_reconnects_when_down_dispatch_is_dropped(self):
+        # on_down() marks the host down but the executor refuses the DOWN
+        # callback, so nothing else will reconnect the control connection.
+        host = self.cluster.metadata.get_host_by_host_id('uuid1')
+        host.set_up()
+        self._use_cluster_down_handling()
+        self.cluster.on_down_potentially_blocking = \
+            Cluster.on_down_potentially_blocking.__get__(self.cluster)
+        self.connection.is_defunct = True
+        self.connection.last_error = ConnectionException(
+            'control connection failed')
+        self.cluster.executor.reset_mock()
+        self.cluster.executor.submit.side_effect = [
+            RuntimeError('cannot schedule new futures'), Mock()]
+
+        self.control_connection._signal_error()
+
+        assert host.is_up is False
+        assert self.cluster.executor.submit.call_args_list[-1] == call(
+            self.control_connection._reconnect)
+
     def test_refresh_network_local_preserves_known_unix_endpoint(self):
         maintenance_endpoint, local_host = \
             self._discover_local_host_over_unix()
@@ -481,51 +1312,59 @@ def test_refresh_network_local_preserves_known_unix_endpoint(self):
         assert host_index[local_host] is not None
         assert Cluster.get_control_connection_host(self.cluster) is local_host
 
-        connection_error = ConnectionException('control connection failed')
+        # A DOWN transition discounted because a usable session pool remains
+        # queues no control on_down callback, so the reconnect is direct.
+        self._discount_down_for(local_host)
         self.connection.is_defunct = True
-        self.connection.last_error = connection_error
-        # Model a conviction whose DOWN transition is discounted because a
-        # usable session pool remains: no control on_down callback is queued.
-        self.cluster.signal_connection_failure = Mock(return_value=True)
+        self.connection.last_error = ConnectionException(
+            'control connection failed')
         self.cluster.executor.reset_mock()
 
         self.control_connection._signal_error()
 
-        self.cluster.signal_connection_failure.assert_called_once_with(
-            local_host, connection_error, is_host_addition=False)
+        assert local_host.is_up is True
+        self.cluster.on_down_potentially_blocking.assert_not_called()
         self.cluster.executor.submit.assert_called_once_with(
             self.control_connection._reconnect)
 
     def test_unix_signal_error_reconnects_if_down_notification_suppressed(self):
         _, local_host = self._discover_local_host_over_unix()
-        connection_error = ConnectionException('control connection failed')
+        session = self._discount_down_for(local_host)
         self.connection.is_defunct = True
-        self.connection.last_error = connection_error
-        self.cluster.signal_connection_failure = Mock(return_value=True)
+        self.connection.last_error = ConnectionException(
+            'control connection failed')
         self.cluster.executor.reset_mock()
 
         self.control_connection._signal_error()
 
-        self.cluster.signal_connection_failure.assert_called_once_with(
-            local_host, connection_error, is_host_addition=False)
+        # The discount is what suppresses the notification here, which means
+        # the host really was resolved from the Unix endpoint: an unresolved
+        # host would reconnect without ever consulting a session pool.
+        session.get_pool_state.assert_called_once_with()
+        assert local_host.is_up is True
+        self.cluster.on_down_potentially_blocking.assert_not_called()
         self.cluster.executor.submit.assert_called_once_with(
             self.control_connection._reconnect)
 
     def test_tcp_route_mismatch_reconnects_if_down_notification_suppressed(self):
         self.control_connection.refresh_node_list_and_token_map()
         local_host = self.cluster.metadata.get_host_by_host_id('uuid1')
+        local_host.set_up()
         self.connection.endpoint = DefaultEndPoint('192.168.1.0', 19042)
         self.connection.original_endpoint = local_host.endpoint
-        connection_error = ConnectionException('control connection failed')
+        session = self._discount_down_for(local_host)
         self.connection.is_defunct = True
-        self.connection.last_error = connection_error
-        self.cluster.signal_connection_failure = Mock(return_value=True)
+        self.connection.last_error = ConnectionException(
+            'control connection failed')
         self.cluster.executor.reset_mock()
 
         self.control_connection._signal_error()
 
-        self.cluster.signal_connection_failure.assert_called_once_with(
-            local_host, connection_error, is_host_addition=False)
+        # As above: reaching the discount proves the host was resolved over
+        # the original endpoint despite the connection's route mismatch.
+        session.get_pool_state.assert_called_once_with()
+        assert local_host.is_up is True
+        self.cluster.on_down_potentially_blocking.assert_not_called()
         self.cluster.executor.submit.assert_called_once_with(
             self.control_connection._reconnect)
 
@@ -561,39 +1400,33 @@ def test_route_mismatch_signal_error_reconnects_if_host_already_down(self):
         _, local_host = self._discover_local_host_over_unix()
         self._refresh_control_connection_over_network()
         local_host.set_down()
-        connection_error = ConnectionException('control connection failed')
+        self._use_cluster_down_handling()
         self.connection.is_defunct = True
-        self.connection.last_error = connection_error
-        self.cluster.signal_connection_failure = Mock(return_value=True)
+        self.connection.last_error = ConnectionException(
+            'control connection failed')
         self.cluster.executor.reset_mock()
 
         self.control_connection._signal_error()
 
-        self.cluster.signal_connection_failure.assert_called_once_with(
-            local_host, connection_error, is_host_addition=False)
+        self.cluster.on_down_potentially_blocking.assert_not_called()
         self.cluster.executor.submit.assert_called_once_with(
             self.control_connection._reconnect)
 
     def test_route_mismatch_signal_error_reconnects_if_host_reconnecting(self):
         _, local_host = self._discover_local_host_over_unix()
         self._refresh_control_connection_over_network()
+        local_host.set_down()
         local_host.get_and_set_reconnection_handler(Mock())
-        connection_error = ConnectionException('control connection failed')
+        self._use_cluster_down_handling()
         self.connection.is_defunct = True
-        self.connection.last_error = connection_error
-
-        def transition_without_notification(host, *_args, **_kwargs):
-            host.set_down()
-            return True
-
-        self.cluster.signal_connection_failure = Mock(
-            side_effect=transition_without_notification)
+        self.connection.last_error = ConnectionException(
+            'control connection failed')
         self.cluster.executor.reset_mock()
 
         self.control_connection._signal_error()
 
-        self.cluster.signal_connection_failure.assert_called_once_with(
-            local_host, connection_error, is_host_addition=False)
+        assert local_host.is_up is False
+        self.cluster.on_down_potentially_blocking.assert_not_called()
         self.cluster.executor.submit.assert_called_once_with(
             self.control_connection._reconnect)