From 59a21c0fe59e795e61dee191ef6cff40e38253ff Mon Sep 17 00:00:00 2001
From: Dmitry Kropachev <dmitry.kropachev@gmail.com>
Date: Tue, 22 Sep 2026 20:07:16 -0400
Subject: [PATCH] reconnection: release stopped host handlers

Authentication failures and exhausted retry schedules left their handlers installed, so every later DOWN event treated a stopped handler as active and permanently skipped the host.

Release terminal handlers without clearing a concurrent replacement, let later DOWN events start a fresh retry cycle, and serialize pending lifecycle transitions so stale restart work cannot replace newer recovery state.

Fixes #1026
---
 CHANGELOG.rst              |   7 +
 cassandra/cluster.py       | 105 ++++++++--
 cassandra/pool.py          |  26 ++-
 tests/unit/test_cluster.py | 384 ++++++++++++++++++++++++++++++++++++-
 4 files changed, 501 insertions(+), 21 deletions(-)

diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index bd2f2b28be..9f062ae2c6 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -29,6 +29,13 @@ Features
   statements skip re-sending result metadata on EXECUTE, and the driver automatically
   refreshes cached metadata when the server detects a schema change (DRIVER-153)
 
+Bug Fixes
+---------
+* A host reconnection handler now releases the host's reconnection slot when
+  authentication fails or its retry schedule is exhausted. A later DOWN event can
+  therefore start a new handler after credentials recover or another reconnection
+  opportunity appears, instead of treating the stopped handler as an active one (#1026).
+
 Others
 ------
 * ``DCAwareRoundRobinPolicy.local_dc`` is now read-only. It is set by the constructor,
diff --git a/cassandra/cluster.py b/cassandra/cluster.py
index a87b2d2082..796b3401ed 100644
--- a/cassandra/cluster.py
+++ b/cassandra/cluster.py
@@ -194,6 +194,7 @@ def new_f(self, *args, **kwargs):
         try:
             future = self.executor.submit(f, self, *args, **kwargs)
             future.add_done_callback(_future_completed)
+            return future
         except Exception:
             log.exception("Failed to submit task to executor")
 
@@ -1903,7 +1904,9 @@ def _on_up_future_completed(self, host, futures, results, lock, finished_future)
 
             log.info("Connection pools established for node %s", host)
             # mark the host as up and notify all listeners
-            host.set_up()
+            with host.lock:
+                host.set_up()
+                host._pending_host_addition = False
             for listener in self.listeners:
                 listener.on_up(host)
         finally:
@@ -1981,6 +1984,7 @@ def on_up(self, host):
             if not have_future:
                 with host.lock:
                     host.set_up()
+                    host._pending_host_addition = False
                     host._currently_handling_node_up = False
 
         # for testing purposes
@@ -1997,12 +2001,15 @@ def _start_reconnector(self, host, is_host_addition):
         # of the current Cluster attributes to create new Connections with
         conn_factory = self._make_connection_factory(host)
 
-        reconnector = _HostReconnectionHandler(
-            host, conn_factory, is_host_addition, self.on_add, self.on_up,
-            self.scheduler, schedule, host.get_and_set_reconnection_handler,
-            new_handler=None)
-
-        old_reconnector = host.get_and_set_reconnection_handler(reconnector)
+        with host.lock:
+            if is_host_addition:
+                host._pending_host_addition = True
+            is_host_addition = host._pending_host_addition
+            reconnector = _HostReconnectionHandler(
+                host, conn_factory, is_host_addition, self.on_add, self.on_up,
+                self.scheduler, schedule,
+                host.get_and_set_reconnection_handler, new_handler=None)
+            old_reconnector = host.get_and_set_reconnection_handler(reconnector)
         if old_reconnector:
             log.debug("Old host reconnector found for %s, cancelling", host)
             old_reconnector.cancel()
@@ -2011,16 +2018,46 @@ def _start_reconnector(self, host, is_host_addition):
         reconnector.start()
 
     @run_in_executor
-    def on_down_potentially_blocking(self, host, is_host_addition):
-        self.profile_manager.on_down(host)
-        self.control_connection.on_down(host)
-        for session in tuple(self.sessions):
-            session.on_down(host)
+    def on_down_potentially_blocking(self, host, is_host_addition,
+                                     down_event_generation):
+        try:
+            with host.lock:
+                if down_event_generation != host._down_event_generation:
+                    return
 
-        for listener in self.listeners:
-            listener.on_down(host)
+            self.profile_manager.on_down(host)
+            self.control_connection.on_down(host)
+            for session in tuple(self.sessions):
+                session.on_down(host)
+
+            for listener in self.listeners:
+                listener.on_down(host)
 
-        self._start_reconnector(host, is_host_addition)
+            self._start_reconnector(host, is_host_addition)
+        finally:
+            with host.lock:
+                if down_event_generation == host._down_event_generation:
+                    host._currently_handling_node_down = False
+
+    @run_in_executor
+    def _restart_reconnector(self, host, is_host_addition,
+                             down_event_generation):
+        try:
+            # Match the old synchronous path's ordering with on_up(): if the
+            # host recovered or is still being brought up while this task was
+            # queued, there is nothing to restart. A failed UP transition
+            # starts its own reconnector before releasing the UP guard, which
+            # must not be replaced by this stale task.
+            with host.lock:
+                if (down_event_generation == host._down_event_generation and
+                        host.is_up is False and
+                        not host._currently_handling_node_up and
+                        not host.is_currently_reconnecting()):
+                    self._start_reconnector(host, is_host_addition)
+        finally:
+            with host.lock:
+                if down_event_generation == host._down_event_generation:
+                    host._currently_handling_node_down = False
 
     def on_down(self, host, is_host_addition, expect_host_to_be_down=False):
         """
@@ -2029,6 +2066,7 @@ def on_down(self, host, is_host_addition, expect_host_to_be_down=False):
         if self.is_shutdown or self.allow_control_connection_query_fallback == ControlConnectionQueryFallback.SkipPoolCreation:
             return
 
+        restart_reconnector = False
         with host.lock:
             was_up = host.is_up
 
@@ -2045,16 +2083,46 @@ def on_down(self, host, is_host_addition, expect_host_to_be_down=False):
                     return
 
             host.set_down()
-            if (not was_up and not expect_host_to_be_down) or host.is_currently_reconnecting():
+            if (not was_up and
+                    (host.is_currently_reconnecting() or
+                     host._currently_handling_node_down)):
                 return
+
+            if not was_up and not expect_host_to_be_down:
+                # A terminal failure may have released this down host's old
+                # handler. Start a new retry cycle without repeating the DOWN
+                # notifications that were sent on the original transition.
+                restart_reconnector = True
+
+            host._currently_handling_node_down = True
+            host._down_event_generation += 1
+            down_event_generation = host._down_event_generation
+
+        if restart_reconnector:
+            future = self._restart_reconnector(
+                host, is_host_addition, down_event_generation)
+            if future is None:
+                with host.lock:
+                    if down_event_generation == host._down_event_generation:
+                        host._currently_handling_node_down = False
+            return
+
         log.warning("Host %s has been marked down", host)
 
-        self.on_down_potentially_blocking(host, is_host_addition)
+        future = self.on_down_potentially_blocking(
+            host, is_host_addition, down_event_generation)
+        if future is None:
+            with host.lock:
+                if down_event_generation == host._down_event_generation:
+                    host._currently_handling_node_down = False
 
     def on_add(self, host, refresh_nodes=True):
         if self.is_shutdown:
             return
 
+        with host.lock:
+            host._pending_host_addition = True
+
         log.debug("Handling new host %r and notifying listeners", host)
 
         self.profile_manager.on_add(host)
@@ -2117,6 +2185,9 @@ def _finalize_add(self, host, set_up=True):
         for listener in self.listeners:
             listener.on_add(host)
 
+        with host.lock:
+            host._pending_host_addition = False
+
         # see if there are any pools to add or remove now that the host is marked up
         for session in tuple(self.sessions):
             session.update_created_pools()
diff --git a/cassandra/pool.py b/cassandra/pool.py
index 14829ffa26..2fb21c9472 100644
--- a/cassandra/pool.py
+++ b/cassandra/pool.py
@@ -161,6 +161,9 @@ class Host(object):
     lock = None
 
     _currently_handling_node_up = False
+    _currently_handling_node_down = False
+    _down_event_generation = 0
+    _pending_host_addition = False
 
     sharding_info = None
 
@@ -355,6 +358,18 @@ def __init__(self, host, connection_factory, is_host_addition, on_add, on_up, *a
         self.host = host
         self.connection_factory = connection_factory
 
+    def start(self):
+        try:
+            _ReconnectionHandler.start(self)
+        except StopIteration:
+            # An empty schedule means this handler will never run.
+            self._release_slot()
+
+    def _release_slot(self):
+        with self.host.lock:
+            if self.host._reconnection_handler is self:
+                self.host._reconnection_handler = None
+
     def try_reconnect(self):
         return self.connection_factory()
 
@@ -367,12 +382,19 @@ def on_reconnection(self, connection):
 
     def on_exception(self, exc, next_delay):
         if isinstance(exc, AuthenticationFailed):
-            return False
+            keep_retrying = False
         else:
             log.warning("Error attempting to reconnect to %s, scheduling retry in %s seconds: %s",
                         self.host, next_delay, exc)
             log.debug("Reconnection error details", exc_info=True)
-            return True
+            keep_retrying = True
+
+        if not keep_retrying or next_delay is None:
+            # This handler will never run again. Release the slot it occupies,
+            # or a later DOWN event will mistake it for a live reconnector.
+            self._release_slot()
+
+        return keep_retrying
 
 
 class HostConnection(object):
diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py
index 74ed346c68..3e585b0be8 100644
--- a/tests/unit/test_cluster.py
+++ b/tests/unit/test_cluster.py
@@ -27,8 +27,8 @@
     ExecutionProfile, _ConfigMode, EXEC_PROFILE_DEFAULT
 from cassandra.connection import ConnectionBusy, ConnectionException
 from cassandra.driver_config import DriverConfigReporter
-from cassandra.pool import Host
-from cassandra.policies import HostDistance, RetryPolicy, RoundRobinPolicy, DowngradingConsistencyRetryPolicy, SimpleConvictionPolicy
+from cassandra.pool import Host, _HostReconnectionHandler
+from cassandra.policies import ExponentialReconnectionPolicy, HostDistance, RetryPolicy, RoundRobinPolicy, DowngradingConsistencyRetryPolicy, SimpleConvictionPolicy
 from cassandra.query import SimpleStatement, named_tuple_factory, tuple_factory
 from tests.unit.utils import mock_session_pools
 from tests import connection_class
@@ -421,6 +421,386 @@ def test_connection_factory_ignores_a_caller_supplied_session_id_and_reporter(se
         assert factory.call_args.kwargs['driver_config_reporter'] is None
 
 
+class HostReconnectionHandlerTest(unittest.TestCase):
+
+    def setUp(self):
+        self.host = Host(
+            "127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4())
+
+    def make_handler(self, connection_factory=None, schedule=None, scheduler=None):
+        handler = _HostReconnectionHandler(
+            self.host, connection_factory or Mock(), False, Mock(), Mock(),
+            scheduler or Mock(), schedule if schedule is not None else iter(()),
+            self.host.get_and_set_reconnection_handler, new_handler=None)
+        self.host.get_and_set_reconnection_handler(handler)
+        return handler
+
+    def test_releases_slot_when_it_gives_up(self):
+        cases = (
+            (AuthenticationFailed('bad credentials'), iter((0, 1.0))),
+            (ConnectionException('refused'), iter((0,))),
+        )
+        for exc, schedule in cases:
+            with self.subTest(exc=exc):
+                handler = self.make_handler(
+                    connection_factory=Mock(side_effect=exc),
+                    schedule=schedule)
+
+                handler.start()
+                handler.run()
+
+                assert not self.host.is_currently_reconnecting()
+
+    def test_keeps_slot_while_retrying(self):
+        scheduler = Mock()
+        handler = self.make_handler(
+            connection_factory=Mock(side_effect=ConnectionException('refused')),
+            schedule=iter((0, 1.0)), scheduler=scheduler)
+
+        handler.start()
+        handler.run()
+
+        assert self.host._reconnection_handler is handler
+        assert scheduler.schedule.call_count == 2
+
+    def test_releases_slot_when_initial_schedule_is_empty(self):
+        scheduler = Mock()
+        schedule = ExponentialReconnectionPolicy(
+            1.0, 60.0, max_attempts=0).new_schedule()
+        handler = self.make_handler(schedule=schedule, scheduler=scheduler)
+
+        handler.start()
+
+        assert not self.host.is_currently_reconnecting()
+        scheduler.schedule.assert_not_called()
+
+    def test_never_releases_replacement(self):
+        handler = self.make_handler()
+        replacement = self.make_handler()
+
+        handler.on_exception(AuthenticationFailed('bad credentials'), 1.0)
+
+        assert self.host._reconnection_handler is replacement
+
+    def test_later_down_event_restarts_after_terminal_failure(self):
+        cluster = Cluster()
+        cluster.scheduler.shutdown()
+        cluster.scheduler = Mock()
+        cluster.executor.shutdown()
+        cluster.executor = Mock()
+        self.addCleanup(cluster.shutdown)
+        cluster._discount_down_events = False
+        cluster.reconnection_policy = Mock()
+        cluster.reconnection_policy.new_schedule.side_effect = (
+            iter((0, 1.0)), iter((0, 1.0)))
+        connection_factory = Mock(
+            side_effect=AuthenticationFailed('bad credentials'))
+        cluster._make_connection_factory = Mock(
+            return_value=connection_factory)
+        cluster.profile_manager.distance = Mock(
+            return_value=HostDistance.LOCAL)
+        cluster.metadata.add_or_return_host(self.host)
+        self.host.set_down()
+
+        cluster._start_reconnector(self.host, is_host_addition=False)
+        stopped_handler = self.host._reconnection_handler
+        stopped_handler.run()
+        assert not self.host.is_currently_reconnecting()
+
+        pending_future = Future()
+        cluster.executor.submit.return_value = pending_future
+        cluster.on_down(self.host, is_host_addition=False)
+
+        assert self.host._reconnection_handler is None
+        assert self.host._currently_handling_node_down
+        cluster.on_down(self.host, is_host_addition=False)
+        cluster.executor.submit.assert_called_once()
+
+        restart_task, *args = cluster.executor.submit.call_args.args
+        restart_task(*args)
+
+        assert self.host._reconnection_handler is not stopped_handler
+        assert self.host.is_currently_reconnecting()
+        assert not self.host._currently_handling_node_down
+
+    def test_later_down_preserves_pending_host_addition(self):
+        cluster = Cluster()
+        cluster.scheduler.shutdown()
+        cluster.scheduler = Mock()
+        cluster.executor.shutdown()
+        cluster.executor = Mock()
+        self.addCleanup(cluster.shutdown)
+        cluster._discount_down_events = False
+        cluster.reconnection_policy = Mock()
+        cluster.reconnection_policy.new_schedule.side_effect = (
+            iter((0, 1.0)), iter((0, 1.0)))
+        connection = Mock()
+        connection_factory = Mock(side_effect=(
+            AuthenticationFailed('bad credentials'), connection))
+        cluster._make_connection_factory = Mock(
+            return_value=connection_factory)
+        cluster.profile_manager.distance = Mock(
+            return_value=HostDistance.LOCAL)
+        cluster.profile_manager.on_add = Mock()
+        cluster.control_connection.on_add = Mock()
+        cluster._prepare_all_queries = Mock()
+        cluster.on_up = Mock()
+        listener = Mock()
+        cluster.register_listener(listener)
+        cluster.metadata.add_or_return_host(self.host)
+        self.host.set_down()
+
+        cluster._start_reconnector(self.host, is_host_addition=True)
+        stopped_handler = self.host._reconnection_handler
+        stopped_handler.run()
+        assert not self.host.is_currently_reconnecting()
+
+        cluster.executor.submit.return_value = Future()
+        cluster.on_down(self.host, is_host_addition=False)
+        restart_task, *args = cluster.executor.submit.call_args.args
+        restart_task(*args)
+        replacement_handler = self.host._reconnection_handler
+        replacement_handler.run()
+
+        assert replacement_handler.is_host_addition
+        listener.on_add.assert_called_once_with(self.host)
+        cluster.on_up.assert_not_called()
+        assert self.host.is_up
+        assert not self.host._pending_host_addition
+
+    def test_queued_reconnector_restart_skips_recovered_host(self):
+        cluster = Cluster()
+        cluster.executor.shutdown()
+        cluster.executor = Mock()
+        self.addCleanup(cluster.shutdown)
+        cluster._discount_down_events = False
+        cluster.metadata.add_or_return_host(self.host)
+        cluster._start_reconnector = Mock()
+        cluster.executor.submit.return_value = Future()
+        self.host.set_down()
+
+        cluster.on_down(self.host, is_host_addition=False)
+        restart_task, *args = cluster.executor.submit.call_args.args
+        self.host.set_up()
+        restart_task(*args)
+
+        cluster._start_reconnector.assert_not_called()
+        assert not self.host._currently_handling_node_down
+
+    def test_fresh_down_supersedes_queued_reconnector_restart(self):
+        cluster = Cluster()
+        cluster.executor.shutdown()
+        cluster.executor = Mock()
+        self.addCleanup(cluster.shutdown)
+        cluster._discount_down_events = False
+        cluster.metadata.add_or_return_host(self.host)
+        cluster.profile_manager.on_down = Mock()
+        cluster.control_connection.on_down = Mock()
+        cluster._start_reconnector = Mock()
+        session = Mock()
+        listener = Mock()
+        cluster.sessions.add(session)
+        cluster.register_listener(listener)
+        cluster.executor.submit.return_value = Future()
+        self.host.set_down()
+
+        cluster.on_down(self.host, is_host_addition=False)
+        self.host.set_up()
+        cluster.on_down(self.host, is_host_addition=False)
+
+        assert cluster.executor.submit.call_count == 2
+        restart_task, *restart_args = (
+            cluster.executor.submit.call_args_list[0].args)
+        down_task, *down_args = cluster.executor.submit.call_args_list[1].args
+
+        restart_task(*restart_args)
+        assert self.host._currently_handling_node_down
+        cluster._start_reconnector.assert_not_called()
+
+        down_task(*down_args)
+        cluster.profile_manager.on_down.assert_called_once_with(self.host)
+        cluster.control_connection.on_down.assert_called_once_with(self.host)
+        session.on_down.assert_called_once_with(self.host)
+        listener.on_down.assert_called_once_with(self.host)
+        cluster._start_reconnector.assert_called_once_with(self.host, False)
+        assert not self.host._currently_handling_node_down
+
+    def test_queued_reconnector_restart_skips_pending_up(self):
+        cluster = Cluster()
+        cluster.scheduler.shutdown()
+        cluster.scheduler = Mock()
+        cluster.executor.shutdown()
+        cluster.executor = Mock()
+        self.addCleanup(cluster.shutdown)
+        cluster._discount_down_events = False
+        cluster.metadata.add_or_return_host(self.host)
+        cluster.profile_manager.distance = Mock(
+            return_value=HostDistance.LOCAL)
+        cluster.profile_manager.on_up = Mock()
+        cluster.profile_manager.on_down = Mock()
+        cluster.control_connection.on_up = Mock()
+        cluster.control_connection.on_down = Mock()
+        cluster._prepare_all_queries = Mock()
+        session = Mock()
+        pool_future = Future()
+        session.add_or_renew_pool.return_value = pool_future
+        cluster.sessions.add(session)
+        cluster.executor.submit.return_value = Future()
+        self.host.set_down()
+
+        cluster.on_up(self.host)
+        assert self.host._currently_handling_node_up
+
+        cluster.on_down(self.host, is_host_addition=False)
+        restart_task, *args = cluster.executor.submit.call_args.args
+        restart_task(*args)
+
+        assert not self.host.is_currently_reconnecting()
+        assert not self.host._currently_handling_node_down
+
+        pool_future.set_result(True)
+        assert self.host.is_up
+        assert not self.host._currently_handling_node_up
+
+        cluster.on_down(self.host, is_host_addition=False)
+        assert cluster.executor.submit.call_count == 2
+        down_task, *args = cluster.executor.submit.call_args.args
+        down_task(*args)
+
+        cluster.profile_manager.on_down.assert_called_once_with(self.host)
+        cluster.control_connection.on_down.assert_called_once_with(self.host)
+        session.on_down.assert_called_once_with(self.host)
+
+    def test_queued_reconnector_restart_preserves_failed_up_handler(self):
+        cluster = Cluster()
+        cluster.scheduler.shutdown()
+        cluster.scheduler = Mock()
+        cluster.executor.shutdown()
+        cluster.executor = Mock()
+        self.addCleanup(cluster.shutdown)
+        cluster._discount_down_events = False
+        cluster.metadata.add_or_return_host(self.host)
+        cluster.profile_manager.distance = Mock(
+            return_value=HostDistance.LOCAL)
+        cluster.profile_manager.on_up = Mock()
+        cluster.profile_manager.on_down = Mock()
+        cluster.control_connection.on_up = Mock()
+        cluster.control_connection.on_down = Mock()
+        cluster._prepare_all_queries = Mock()
+        cluster._make_connection_factory = Mock(return_value=Mock())
+        session = Mock()
+        pool_future = Future()
+        session.add_or_renew_pool.return_value = pool_future
+        cluster.sessions.add(session)
+        cluster.executor.submit.return_value = Future()
+        self.host.set_down()
+
+        cluster.on_up(self.host)
+        cluster.on_down(self.host, is_host_addition=False)
+        restart_task, *args = cluster.executor.submit.call_args.args
+
+        pool_future.set_result(False)
+        failed_up_handler = self.host._reconnection_handler
+        assert failed_up_handler is not None
+        assert not self.host._currently_handling_node_up
+
+        restart_task(*args)
+
+        assert self.host._reconnection_handler is failed_up_handler
+        assert not failed_up_handler._cancelled
+        assert cluster.scheduler.schedule.call_count == 1
+        assert not self.host._currently_handling_node_down
+
+    def test_successful_up_resolves_pending_host_addition(self):
+        cluster = Cluster()
+        cluster.scheduler.shutdown()
+        cluster.scheduler = Mock()
+        cluster.executor.shutdown()
+        cluster.executor = Mock()
+        self.addCleanup(cluster.shutdown)
+        cluster._discount_down_events = False
+        cluster.metadata.add_or_return_host(self.host)
+        cluster.profile_manager.distance = Mock(
+            return_value=HostDistance.LOCAL)
+        cluster.profile_manager.on_add = Mock()
+        cluster.profile_manager.on_up = Mock()
+        cluster.profile_manager.on_down = Mock()
+        cluster.control_connection.on_add = Mock()
+        cluster.control_connection.on_up = Mock()
+        cluster.control_connection.on_down = Mock()
+        cluster._prepare_all_queries = Mock()
+        cluster._make_connection_factory = Mock(return_value=Mock())
+        listener = Mock()
+        cluster.register_listener(listener)
+        session = Mock()
+        failed_add_future = Future()
+        up_future = Future()
+        session.add_or_renew_pool.side_effect = (
+            failed_add_future, up_future)
+        cluster.sessions.add(session)
+        cluster.executor.submit.return_value = Future()
+
+        cluster.on_add(self.host)
+        failed_add_future.set_result(False)
+        assert self.host._pending_host_addition
+        listener.on_add.assert_not_called()
+
+        cluster.on_up(self.host)
+        up_future.set_result(True)
+
+        assert self.host.is_up
+        assert not self.host._pending_host_addition
+        listener.on_up.assert_called_once_with(self.host)
+
+        cluster.on_down(self.host, is_host_addition=False)
+        down_task, *args = cluster.executor.submit.call_args.args
+        down_task(*args)
+
+        assert not self.host._reconnection_handler.is_host_addition
+        listener.on_add.assert_not_called()
+
+    def test_repeated_down_waits_for_pending_down_processing(self):
+        cluster = Cluster()
+        cluster.scheduler.shutdown()
+        cluster.scheduler = Mock()
+        cluster.executor.shutdown()
+        cluster.executor = Mock()
+        self.addCleanup(cluster.shutdown)
+        cluster._discount_down_events = False
+        cluster.metadata.add_or_return_host(self.host)
+        cluster.profile_manager.on_down = Mock()
+        cluster.control_connection.on_down = Mock()
+        cluster._start_reconnector = Mock()
+        pending_future = Future()
+        cluster.executor.submit.return_value = pending_future
+        self.host.set_up()
+
+        cluster.on_down(self.host, is_host_addition=False)
+        cluster.on_down(self.host, is_host_addition=False)
+
+        cluster.executor.submit.assert_called_once()
+        cluster._start_reconnector.assert_not_called()
+
+        down_task, *args = cluster.executor.submit.call_args.args
+        down_task(*args)
+
+        cluster._start_reconnector.assert_called_once_with(self.host, False)
+        assert not self.host._currently_handling_node_down
+
+    def test_failed_down_submission_releases_pending_state(self):
+        cluster = Cluster()
+        cluster.executor.shutdown()
+        cluster.executor = Mock()
+        self.addCleanup(cluster.shutdown)
+        cluster._discount_down_events = False
+        cluster.metadata.add_or_return_host(self.host)
+        cluster.executor.submit.side_effect = RuntimeError('executor stopped')
+        self.host.set_up()
+
+        cluster.on_down(self.host, is_host_addition=False)
+
+        assert not self.host._currently_handling_node_down
+
 class SchedulerTest(unittest.TestCase):
     # TODO: this suite could be expanded; for now just adding a test covering a ticket