From 3aac367d834e5ccb8bb3cbb0029ffd56c42f0e49 Mon Sep 17 00:00:00 2001
From: Dmitry Kropachev <dmitry.kropachev@gmail.com>
Date: Fri, 23 Oct 2026 19:58:37 -0400
Subject: [PATCH 1/7] pool: fence connection failures against retired ownership

A returned socket can fail after its pool or shard slot has been replaced.
The old failure can convict the host, remove the replacement slot, or shut
healthy sockets down. Reserve the owning session pool, host endpoint, and
connection identity before reporting a failure, and revalidate ownership
at DOWN admission and destructive pool cleanup.

Run conviction callbacks and socket closure outside ownership locks.
Retain replacement ownership per shard so a queued old worker cannot
remove a newer connection or suppress its subsequent recovery. Preserve
stream accounting and weak session ownership for stale completions.

Promote stale-socket regressions from expected failures and cover policy
re-entry, health-inspection races, queued replacements, and current failure
controls, including forced shutdown and local send errors.

Fixes: #858
Refs: #925
Refs: #921
---
 cassandra/_lifecycle.py                   |  94 +++++--
 cassandra/pool.py                         | 104 ++++---
 tests/unit/test_host_connection_pool.py   | 313 ++++++++++++++++------
 tests/unit/test_lifecycle_coordination.py |  69 +++++
 4 files changed, 448 insertions(+), 132 deletions(-)

diff --git a/cassandra/_lifecycle.py b/cassandra/_lifecycle.py
index ccb7910423..a9dab7b932 100644
--- a/cassandra/_lifecycle.py
+++ b/cassandra/_lifecycle.py
@@ -20,7 +20,7 @@
 
 from collections import deque
 from concurrent.futures import Future
-from contextlib import contextmanager, nullcontext
+from contextlib import contextmanager
 from functools import partial, wraps
 import logging
 from threading import local, RLock
@@ -56,9 +56,60 @@ def owns_locked(self):
                     self.host, self.generation, self.expected_pool, self.endpoint))
 
     def is_current(self):
+        with self.locked():
+            return self.owns_locked()
+
+    @contextmanager
+    def locked(self):
+        with self.session._lock:
+            with self.host.lock:
+                yield
+
+
+class ConnectionFailureReservation:
+    """Retain the pool and shard slot that own a connection failure."""
+
+    def __init__(self, pool, connection, replacing=False):
+        self.pool = pool
+        self.connection = connection
+        self.session = pool._session_ref()
+        self.host = pool.host
+        self.replacing = replacing
+
+    @contextmanager
+    def locked(self):
+        if self.session is None:
+            yield
+            return
         with self.session._lock:
             with self.host.lock:
-                return self.owns_locked()
+                # Unpublished candidates can hold their work lock while
+                # publication waits for Session under the membership lock.
+                # Only a current pool needs its shard ownership inspected.
+                if self.session._pools.get(self.host) is not self.pool:
+                    yield
+                    return
+                with self.pool._lock:
+                    yield
+
+    def owns_locked(self):
+        if (self.session is None or
+                self.session._pools.get(self.host) is not self.pool):
+            return False
+        current = self.pool._connections.get(self.connection.features.shard_id)
+        owns_slot = (current is self.connection or
+                     (self.replacing and current is None and
+                      self.pool._replacement_connections.get(
+                          self.connection.features.shard_id) is self.connection))
+        return (not self.session.is_shutdown and
+                not self.session.cluster.is_shutdown and
+                not self.host._is_removed and
+                self.host.endpoint == self.pool.endpoint and
+                not self.pool.is_shutdown and owns_slot)
+
+    def is_current(self):
+        with self.locked():
+            return self.owns_locked()
 
 
 class LifecycleCoordinator:
@@ -196,27 +247,26 @@ def observe_health(self, host):
     def begin_down(self, host, health_generation, connected=False,
                    expect_down=False, pool_failure=None):
         """Return (generation, retry_only), or None if DOWN is discounted."""
-        failure_lock = (pool_failure.session._lock
-                        if pool_failure is not None else nullcontext())
+        failure_lock = (pool_failure.locked()
+                        if pool_failure is not None else host.lock)
         with failure_lock:
-            with host.lock:
-                if (self.closed or host._is_removed or connected or
-                        health_generation != host._health_generation or
-                        (pool_failure is not None and not pool_failure.owns_locked())):
-                    return None
-                host._health_generation += 1
-                self.cancel_preparation(host)
-                host._readiness_generation = (
-                    getattr(host, '_readiness_generation', 0) + 1)
-                host._currently_handling_node_up = False
-                was_up = host.is_up
-                host.set_down()
-                if (not was_up and (host.is_currently_reconnecting() or
-                                    host._currently_handling_node_down)):
-                    return None
-                host._currently_handling_node_down = True
-                host._down_event_generation += 1
-                return host._down_event_generation, not was_up and not expect_down
+            if (self.closed or host._is_removed or connected or
+                    health_generation != host._health_generation or
+                    (pool_failure is not None and not pool_failure.owns_locked())):
+                return None
+            host._health_generation += 1
+            self.cancel_preparation(host)
+            host._readiness_generation = (
+                getattr(host, '_readiness_generation', 0) + 1)
+            host._currently_handling_node_up = False
+            was_up = host.is_up
+            host.set_down()
+            if (not was_up and (host.is_currently_reconnecting() or
+                                host._currently_handling_node_down)):
+                return None
+            host._currently_handling_node_down = True
+            host._down_event_generation += 1
+            return host._down_event_generation, not was_up and not expect_down
 
     def dispatch_down(self, host, health_generation, is_host_addition,
                       connected, expect_host_to_be_down, on_add_reconnection,
diff --git a/cassandra/pool.py b/cassandra/pool.py
index a9c35cc375..6b78126581 100644
--- a/cassandra/pool.py
+++ b/cassandra/pool.py
@@ -30,7 +30,7 @@
 except ImportError:
     from cassandra.util import WeakSet  # NOQA
 
-from cassandra import AuthenticationFailed
+from cassandra import AuthenticationFailed, _lifecycle
 from cassandra.connection import (ConnectionException, ConnectionSendError, EndPoint,
                                   DefaultEndPoint, UnixSocketEndPoint)
 from cassandra.policies import HostDistance
@@ -468,11 +468,12 @@ def __init__(self, host, host_distance, session):
         self.endpoint = host.endpoint
         self.host_distance = host_distance
         self._session = weakref.proxy(session)
+        self._session_ref = weakref.ref(session)
         self._lock = Lock()
         self._membership_lock = Lock()
         # this is used in conjunction with the connection streams. Not using the connection lock because the connection can be replaced in the lifetime of the pool.
         self._stream_available_condition = Condition(Lock())
-        self._is_replacing = False
+        self._replacement_connections = {}
         self._connecting = set()
         self._connections = {}
         self._pending_connections = []
@@ -654,41 +655,62 @@ def return_connection(self, connection, stream_was_orphaned=False,
                 self._stream_available_condition.notify()
 
         if connection.is_defunct or connection.is_closed:
-            if connection.signaled_error and not replace_only and \
-                    (not self.shutdown_on_error or
-                     (isinstance(connection.last_error, ConnectionSendError) and
-                      not heartbeat_failed)):
+            reservation = _lifecycle.ConnectionFailureReservation(self, connection)
+            with reservation.locked():
+                current = reservation.owns_locked()
+                if current:
+                    if connection.signaled_error and not replace_only and \
+                            (not self.shutdown_on_error or
+                             (isinstance(connection.last_error, ConnectionSendError) and
+                              not heartbeat_failed)):
+                        return
+                    report_failure = not replace_only and not connection.signaled_error
+                    connection.signaled_error = True
+            if not current:
+                with self._lock:
+                    self._trash.discard(connection)
+                # A stale completion still releases its stream above, but its
+                # failed transport cannot affect the current pool or shard.
+                connection.close()
                 return
 
             is_down = False
-            if replace_only:
-                # A local send-queue failure says nothing about host health.
-                # Mark the connection handled so later callbacks do not report
-                # this failure to the host conviction policy.
-                connection.signaled_error = True
-            elif not connection.signaled_error:
+            if report_failure:
                 log.debug("Defunct or closed connection (%s) returned to pool, potentially "
                           "marking host %s as down", id(connection), self.host)
                 is_down = self.host.signal_connection_failure(connection.last_error)
-                connection.signaled_error = True
 
             if self.shutdown_on_error and not is_down and not replace_only:
                 is_down = True
 
             if is_down:
-                self.shutdown()
-                self._session.cluster.on_down(self.host, is_host_addition=False)
+                # Policy and health inspection can replace the pool or socket.
+                # Carry ownership through DOWN admission and pool shutdown.
+                if reservation.is_current():
+                    try:
+                        self._session.cluster.on_down(
+                            self.host, is_host_addition=False,
+                            _pool_failure=reservation)
+                    finally:
+                        self.shutdown(_failure=reservation)
+                else:
+                    with self._lock:
+                        self._trash.discard(connection)
+                    connection.close()
             else:
+                schedule_replacement = False
+                with reservation.locked():
+                    if reservation.owns_locked():
+                        with self._membership_lock:
+                            self._connections.pop(connection.features.shard_id)
+                        self._replacement_connections[
+                            connection.features.shard_id] = connection
+                        schedule_replacement = True
+                if not schedule_replacement:
+                    with self._lock:
+                        self._trash.discard(connection)
                 connection.close()
-                with self._lock:
-                    if self.is_shutdown:
-                        return
-                    with self._membership_lock:
-                        self._connections.pop(
-                            connection.features.shard_id, None)
-                    if self._is_replacing:
-                        return
-                    self._is_replacing = True
+                if schedule_replacement:
                     self._session.submit(self._replace, connection)
         elif connection in self._trash:
             with connection.lock:
@@ -712,11 +734,25 @@ def on_orphaned_stream_released(self):
             self._stream_available_condition.notify()
 
     def _replace(self, connection):
+        reservation = _lifecycle.ConnectionFailureReservation(
+            self, connection, replacing=True)
+        if not reservation.is_current():
+            with self._lock:
+                shard_id = connection.features.shard_id
+                if self._replacement_connections.get(shard_id) is connection:
+                    self._replacement_connections.pop(shard_id)
+            return
         failed_connection = None
         with self._lock:
-            if self.is_shutdown:
+            shard_id = connection.features.shard_id
+            current = self._connections.get(shard_id)
+            if (self.is_shutdown or
+                    (current is not None and current is not connection) or
+                    (current is None and
+                     self._replacement_connections.get(shard_id) is not connection)):
                 return
 
+            self._replacement_connections[shard_id] = connection
             log.debug("Replacing connection (%s) to %s", id(connection), self.host)
             try:
                 with self._membership_lock:
@@ -734,17 +770,21 @@ def _replace(self, connection):
                         raise
             except Exception:
                 log.warning("Failed reconnecting %s. Retrying." % (self.endpoint,))
+                self._replacement_connections[shard_id] = connection
                 self._session.submit(self._replace, connection)
             else:
-                self._is_replacing = False
+                self._replacement_connections.pop(shard_id, None)
                 with self._stream_available_condition:
                     self._stream_available_condition.notify()
         if failed_connection is not None:
             failed_connection.close()
 
-    def shutdown(self):
+    def shutdown(self, _failure=None):
         log.debug("Shutting down connections to %s", self.host)
-        with self._lock:
+        shutdown_lock = _failure.locked() if _failure is not None else self._lock
+        with shutdown_lock:
+            if (_failure is not None and not _failure.owns_locked()):
+                return
             with self._membership_lock:
                 if self.is_shutdown:
                     return
@@ -754,12 +794,14 @@ def shutdown(self):
             with self._stream_available_condition:
                 self._stream_available_condition.notify_all()
 
-            for future in self._shard_connections_futures:
-                future.cancel()
-
+            futures_to_cancel = tuple(self._shard_connections_futures)
+            self._replacement_connections.clear()
             pending_connections_to_close = self._pending_connections.copy()
             self._pending_connections.clear()
 
+        for future in futures_to_cancel:
+            future.cancel()
+
         # connection.close can call pool.return_connection, which will
         #  obtain self._lock via self._stream_available_condition.
         # So, it never should be called within self._lock context
diff --git a/tests/unit/test_host_connection_pool.py b/tests/unit/test_host_connection_pool.py
index b93e99bd14..10a520c405 100644
--- a/tests/unit/test_host_connection_pool.py
+++ b/tests/unit/test_host_connection_pool.py
@@ -11,7 +11,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import Future, ThreadPoolExecutor
 import logging
 import uuid
 from cassandra.protocol_features import ProtocolFeatures
@@ -19,10 +19,10 @@
 from cassandra.shard_info import _ShardingInfo
 
 import unittest
-from threading import Thread, Event, Lock
+from threading import Thread, Event, Lock, RLock
 from unittest.mock import Mock, NonCallableMagicMock, MagicMock, patch
 
-from cassandra.cluster import Session, ShardAwareOptions
+from cassandra.cluster import Cluster, Session, ShardAwareOptions
 from cassandra.connection import Connection, ConnectionBusy, ConnectionException, ConnectionSendError, DefaultEndPoint
 from cassandra.pool import HostConnection
 from cassandra.pool import Host, NoConnectionsAvailable
@@ -34,38 +34,12 @@
 LOGGER = logging.getLogger(__name__)
 
 
-class _ExpectedStaleSocketFailure(Exception):
-    pass
-
-
-@pytest.mark.parametrize('ownership, convicted, postcondition', [
-    pytest.param('current', False, None),
-    pytest.param('current', True, None),
-] + [
-    pytest.param(
-        ownership, convicted, postcondition,
-        marks=pytest.mark.xfail(
-            strict=True, raises=_ExpectedStaleSocketFailure,
-            reason='#858: stale socket failures still affect current ownership'
-        ) if postcondition in expected_failures else ())
-    for ownership, convicted, expected_failures in [
-        ('draining', False, ('conviction', 'replacement_scheduled',
-                             'shard_connection')),
-        ('draining', True, ('conviction', 'down', 'pool_open',
-                            'shard_connection', 'replacement_open')),
-        ('retired_pool', False, ('conviction',)),
-        ('retired_pool', True, ('conviction', 'down')),
-    ]
-    for postcondition in [
-        'conviction', 'down', 'replacement_scheduled', 'pool_open',
-        'shard_connection', 'replacement_open',
-    ]
-])
-def test_connection_failure_respects_socket_ownership(
-        ownership, convicted, postcondition):
+def _make_socket_pool():
     host = Host('127.0.0.1', SimpleConvictionPolicy, host_id=uuid.uuid4())
-    host.signal_connection_failure = Mock(return_value=convicted)
-    session = NonCallableMagicMock(spec=Session, keyspace=None)
+    host.signal_connection_failure = Mock(return_value=False)
+    session = NonCallableMagicMock(
+        spec=Session, keyspace=None, _lock=RLock(), is_shutdown=False)
+    session.cluster.is_shutdown = False
     session.cluster.shard_aware_options.disable = True
 
     def connection():
@@ -80,6 +54,25 @@ def connection():
     session.cluster.connection_factory.side_effect = [original, replacement]
     pool = HostConnection(host, HostDistance.LOCAL, session)
     session._pools = {host: pool}
+    return host, session, pool, original, replacement
+
+
+@pytest.fixture
+def socket_pool():
+    return _make_socket_pool()
+
+
+@pytest.mark.parametrize('ownership', [
+    'current', 'draining', 'retired_pool', 'unpublished_pool',
+    'retired_host', 'moved_endpoint', 'shutdown_session',
+])
+@pytest.mark.parametrize('convicted', [False, True])
+@pytest.mark.parametrize('shutdown_on_error', [False, True])
+def test_connection_failure_respects_socket_ownership(
+        socket_pool, ownership, convicted, shutdown_on_error):
+    host, session, pool, original, replacement = socket_pool
+    host.signal_connection_failure.return_value = convicted
+    pool.shutdown_on_error = shutdown_on_error
     borrowed, _ = pool.borrow_connection(timeout=0)
     assert borrowed is original
     assert original.in_flight == 1
@@ -89,10 +82,17 @@ def connection():
         # slot has been given to a healthy replacement.
         pool._connections[0] = replacement
         pool._trash.add(original)
-    elif ownership == 'retired_pool':
+    elif ownership in ('retired_pool', 'unpublished_pool'):
         winner = HostConnection(host, HostDistance.LOCAL, session)
         session._pools = {host: winner}
-        pool.shutdown()
+        if ownership == 'retired_pool':
+            pool.shutdown()
+    elif ownership == 'retired_host':
+        host._is_removed = True
+    elif ownership == 'moved_endpoint':
+        host.endpoint = DefaultEndPoint('127.0.0.2')
+    elif ownership == 'shutdown_session':
+        session.is_shutdown = True
 
     original.is_defunct = True
     original.last_error = ConnectionException('old socket failed', host.endpoint)
@@ -103,29 +103,171 @@ def connection():
     assert replacement.in_flight == 0
     if ownership == 'current':
         host.signal_connection_failure.assert_called_once_with(original.last_error)
-        assert pool.is_shutdown is convicted
-        if convicted:
-            session.cluster.on_down.assert_called_once_with(
-                host, is_host_addition=False)
+        assert pool.is_shutdown is (convicted or shutdown_on_error)
+        if convicted or shutdown_on_error:
+            session.cluster.on_down.assert_called_once()
+            args, kwargs = session.cluster.on_down.call_args
+            assert args == (host,)
+            assert kwargs['is_host_addition'] is False
         else:
             session.submit.assert_called_once_with(pool._replace, original)
     else:
         current = session._pools[host]
+        host.signal_connection_failure.assert_not_called()
+        session.cluster.on_down.assert_not_called()
+        session.submit.assert_not_called()
+        assert not current.is_shutdown
+        if ownership in ('draining', 'retired_pool', 'unpublished_pool'):
+            assert current._connections.get(0) is replacement
+            replacement.close.assert_not_called()
+
+
+@pytest.mark.parametrize('convicted', [False, True])
+@pytest.mark.parametrize('change', ['socket', 'pool', 'endpoint', 'retirement'])
+def test_failure_policy_cannot_affect_replacement(socket_pool, convicted, change):
+    host, session, pool, original, replacement = socket_pool
+    pool.borrow_connection(timeout=0)
+    original.is_defunct = True
+    original.last_error = ConnectionException('socket failed', host.endpoint)
+    changed = Event()
+    errors = []
+
+    def replace():
         try:
-            if postcondition == 'conviction':
-                host.signal_connection_failure.assert_not_called()
-            elif postcondition == 'down':
-                session.cluster.on_down.assert_not_called()
-            elif postcondition == 'replacement_scheduled':
-                session.submit.assert_not_called()
-            elif postcondition == 'pool_open':
-                assert not current.is_shutdown
-            elif postcondition == 'shard_connection':
-                assert current._connections.get(0) is replacement
-            elif postcondition == 'replacement_open':
-                replacement.close.assert_not_called()
-        except AssertionError as error:
-            raise _ExpectedStaleSocketFailure from error
+            winner = (HostConnection(host, HostDistance.LOCAL, session)
+                      if change == 'pool' else None)
+            with session._lock, host.lock, pool._lock:
+                if change == 'socket':
+                    pool._connections[0] = replacement
+                    pool._trash.add(original)
+                elif change == 'pool':
+                    session._pools[host] = winner
+                elif change == 'endpoint':
+                    host.endpoint = DefaultEndPoint('127.0.0.2')
+                else:
+                    host._is_removed = True
+        except Exception as error:
+            errors.append(error)
+        finally:
+            changed.set()
+
+    worker = Thread(target=replace)
+
+    def convict(error):
+        worker.start()
+        assert changed.wait(5), 'policy ran while ownership locks were held'
+        assert not errors
+        return convicted
+
+    host.signal_connection_failure.side_effect = convict
+    try:
+        pool.return_connection(original)
+    finally:
+        worker.join(5)
+    assert not worker.is_alive()
+    assert original.in_flight == 0
+    session.cluster.on_down.assert_not_called()
+    session.submit.assert_not_called()
+    assert not session._pools[host].is_shutdown
+    if change in ('socket', 'pool'):
+        assert session._pools[host]._connections[0] is replacement
+        replacement.close.assert_not_called()
+
+
+@pytest.mark.parametrize('change', ['socket', 'pool'])
+def test_down_admission_rechecks_failed_socket(socket_pool, change):
+    host, session, pool, original, replacement = socket_pool
+    cluster = Cluster(contact_points=[])
+    session.cluster = cluster
+    cluster.sessions = (session,)
+    cluster.profile_manager = Mock()
+    cluster.on_down_potentially_blocking = Mock(return_value=Future())
+    host.set_up()
+    host.signal_connection_failure.return_value = True
+    pool.borrow_connection(timeout=0)
+    original.is_defunct = True
+    original.last_error = ConnectionException('socket failed', host.endpoint)
+
+    def distance(current):
+        with session._lock, host.lock, pool._lock:
+            if change == 'socket':
+                pool._connections[0] = replacement
+            else:
+                session._pools[host] = Mock(
+                    host=host, endpoint=host.endpoint, is_shutdown=False)
+        return HostDistance.IGNORED
+
+    cluster.profile_manager.distance.side_effect = distance
+    try:
+        pool.return_connection(original)
+        assert host.is_up
+        assert not session._pools[host].is_shutdown
+        replacement.close.assert_not_called()
+        cluster.on_down_potentially_blocking.assert_not_called()
+    finally:
+        pool.shutdown()
+        cluster.shutdown()
+
+
+def test_queued_replacement_cannot_remove_new_socket(socket_pool):
+    host, session, pool, original, replacement = socket_pool
+    pool.borrow_connection(timeout=0)
+    original.is_defunct = True
+    pool.return_connection(original)
+    session.submit.assert_called_once_with(pool._replace, original)
+
+    with pool._lock:
+        pool._connections[0] = replacement
+    pool._replace(original)
+    assert pool._connections[0] is replacement
+    assert session.cluster.connection_factory.call_count == 1
+    replacement.close.assert_not_called()
+
+    # The stale worker must neither suppress a new failure nor clear its owner.
+    pool.borrow_connection(timeout=0)
+    replacement.is_defunct = True
+    pool.return_connection(replacement)
+    session.submit.assert_called_with(pool._replace, replacement)
+    pool._replace(original)
+    healthy = HashableMock(spec=Connection)
+    healthy.features = ProtocolFeatures(shard_id=0)
+    session.cluster.connection_factory.side_effect = [healthy]
+    pool._replace(replacement)
+    assert pool._connections[0] is healthy
+
+
+def test_current_pool_still_shuts_down_when_down_dispatch_raises(socket_pool):
+    host, session, pool, original, _ = socket_pool
+    host.signal_connection_failure.return_value = True
+    session.cluster.on_down.side_effect = RuntimeError('executor stopped')
+    pool.borrow_connection(timeout=0)
+    original.is_defunct = True
+
+    with pytest.raises(RuntimeError, match='executor stopped'):
+        pool.return_connection(original)
+
+    assert pool.is_shutdown
+    assert original.in_flight == 0
+    original.close.assert_called_once()
+
+
+def test_return_after_session_collection_releases_original_stream():
+    import gc
+    import weakref
+
+    host, session, pool, original, _ = _make_socket_pool()
+    pool.borrow_connection(timeout=0)
+    original.is_defunct = True
+    session_ref = weakref.ref(session)
+    del session
+    gc.collect()
+    assert session_ref() is None
+
+    pool.return_connection(original)
+
+    assert original.in_flight == 0
+    original.close.assert_called_once()
+    host.signal_connection_failure.assert_not_called()
 
 
 class _PoolTests(unittest.TestCase):
@@ -134,16 +276,26 @@ class _PoolTests(unittest.TestCase):
     uses_single_connection = None
 
     def make_session(self):
-        session = NonCallableMagicMock(spec=Session, keyspace='foobarkeyspace', _trash=[])
+        session = NonCallableMagicMock(
+            spec=Session, keyspace='foobarkeyspace', _trash=[],
+            _lock=RLock(), _pools={}, is_shutdown=False)
+        session.cluster.is_shutdown = False
         return session
 
+    def make_pool(self, host, session):
+        host.lock = RLock()
+        host._is_removed = False
+        pool = self.PoolImpl(host, HostDistance.LOCAL, session)
+        session._pools[host] = pool
+        return pool
+
     def test_borrow_and_return(self):
         host = Mock(spec=Host, address='ip1')
         session = self.make_session()
         conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False, is_closed=False, max_request_id=100)
         session.cluster.connection_factory.return_value = conn
 
-        pool = self.PoolImpl(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         session.cluster.connection_factory.assert_called_once_with(host.endpoint, on_orphaned_stream_released=pool.on_orphaned_stream_released)
 
         c, request_id = pool.borrow_connection(timeout=0.01)
@@ -165,7 +317,7 @@ def test_init_logs_none_shard_id(self):
         session.cluster.connection_factory.return_value = conn
 
         with self.assertLogs('cassandra.pool', level='DEBUG') as logs:
-            self.PoolImpl(host, HostDistance.LOCAL, session)
+            self.make_pool(host, session)
 
         shard_logs = [r.getMessage() for r in logs.records
                       if "First connection created" in r.getMessage()]
@@ -178,7 +330,7 @@ def test_failed_wait_for_connection(self):
         conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False, is_closed=False, max_request_id=100)
         session.cluster.connection_factory.return_value = conn
 
-        pool = self.PoolImpl(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         session.cluster.connection_factory.assert_called_once_with(host.endpoint, on_orphaned_stream_released=pool.on_orphaned_stream_released)
 
         pool.borrow_connection(timeout=0.01)
@@ -198,7 +350,7 @@ def test_successful_wait_for_connection(self):
                                     lock=Lock())
         session.cluster.connection_factory.return_value = conn
 
-        pool = self.PoolImpl(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         session.cluster.connection_factory.assert_called_once_with(host.endpoint, on_orphaned_stream_released=pool.on_orphaned_stream_released)
 
         pool.borrow_connection(timeout=0.01)
@@ -223,7 +375,7 @@ def test_spawn_when_at_max(self):
         conn.max_request_id = 100
         session.cluster.connection_factory.return_value = conn
 
-        pool = self.PoolImpl(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         session.cluster.connection_factory.assert_called_once_with(host.endpoint, on_orphaned_stream_released=pool.on_orphaned_stream_released)
 
         pool.borrow_connection(timeout=0.01)
@@ -247,7 +399,7 @@ def test_return_defunct_connection(self):
                                     max_request_id=100, signaled_error=False)
         session.cluster.connection_factory.return_value = conn
 
-        pool = self.PoolImpl(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         session.cluster.connection_factory.assert_called_once_with(host.endpoint, on_orphaned_stream_released=pool.on_orphaned_stream_released)
 
         pool.borrow_connection(timeout=0.01)
@@ -267,7 +419,7 @@ def test_replace_only_does_not_mark_host_down(self):
                             is_closed=False, max_request_id=100,
                             signaled_error=False)
         session.cluster.connection_factory.return_value = conn
-        pool = self.PoolImpl(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
 
         pool.borrow_connection(timeout=0.01)
         conn.is_defunct = True
@@ -286,7 +438,7 @@ def test_signaled_error_still_shuts_down_when_configured(self):
                             is_closed=False, max_request_id=100,
                             signaled_error=False)
         session.cluster.connection_factory.return_value = conn
-        pool = self.PoolImpl(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         pool.shutdown_on_error = True
 
         pool.borrow_connection(timeout=0.01)
@@ -304,7 +456,7 @@ def test_local_send_error_skips_shutdown_when_configured(self):
                             is_closed=False, max_request_id=100,
                             signaled_error=False)
         session.cluster.connection_factory.return_value = conn
-        pool = self.PoolImpl(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         pool.shutdown_on_error = True
 
         pool.borrow_connection(timeout=0.01)
@@ -326,7 +478,7 @@ def test_heartbeat_failure_shuts_down_after_local_send_error(self):
                             is_closed=False, max_request_id=100,
                             signaled_error=False)
         session.cluster.connection_factory.return_value = conn
-        pool = self.PoolImpl(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         pool.shutdown_on_error = True
 
         pool.borrow_connection(timeout=0.01)
@@ -337,7 +489,8 @@ def test_heartbeat_failure_shuts_down_after_local_send_error(self):
 
         assert pool.is_shutdown
         host.signal_connection_failure.assert_not_called()
-        session.cluster.on_down.assert_called_once_with(host, is_host_addition=False)
+        session.cluster.on_down.assert_called_once_with(
+            host, is_host_addition=False, _pool_failure=unittest.mock.ANY)
 
     def test_return_defunct_connection_on_down_host(self):
         host = Mock(spec=Host, address='ip1')
@@ -348,7 +501,7 @@ def test_return_defunct_connection_on_down_host(self):
         session.cluster.connection_factory.return_value = conn
         session.cluster.shard_aware_options = ShardAwareOptions()
 
-        pool = self.PoolImpl(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         session.cluster.connection_factory.assert_called_once_with(host.endpoint, on_orphaned_stream_released=pool.on_orphaned_stream_released)
 
         pool.borrow_connection(timeout=0.01)
@@ -375,7 +528,7 @@ def test_return_closed_connection(self):
                                     signaled_error=False, orphaned_threshold_reached=False)
         session.cluster.connection_factory.return_value = conn
 
-        pool = self.PoolImpl(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         session.cluster.connection_factory.assert_called_once_with(host.endpoint, on_orphaned_stream_released=pool.on_orphaned_stream_released)
 
         pool.borrow_connection(timeout=0.01)
@@ -471,7 +624,7 @@ class HostConnectionTests(_PoolTests):
     PoolImpl = HostConnection
     uses_single_connection = True
 
-    def test_replacement_connection_uses_pool_endpoint(self):
+    def test_replacement_skips_pool_after_endpoint_changes(self):
         host = Host('127.0.0.1', SimpleConvictionPolicy, host_id=uuid.uuid4())
         session = self.make_session()
         session.keyspace = None
@@ -483,14 +636,16 @@ def test_replacement_connection_uses_pool_endpoint(self):
         replacement = HashableMock(spec=Connection)
         replacement.features.shard_id = 0
         session.cluster.connection_factory.side_effect = [first, replacement]
-        pool = HostConnection(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
 
         original_endpoint = host.endpoint
         host.endpoint = DefaultEndPoint('127.0.0.2')
         pool._replace(first)
 
         assert pool.endpoint == original_endpoint
-        assert session.cluster.connection_factory.call_args.args[0] == original_endpoint
+        assert session.cluster.connection_factory.call_count == 1
+        assert pool._connections[0] is first
+        replacement.close.assert_not_called()
         pool.shutdown()
 
     def test_borrow_rejects_unsynchronized_connection_without_waiting(self):
@@ -499,7 +654,7 @@ def test_borrow_rejects_unsynchronized_connection_without_waiting(self):
         conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False,
                             is_closed=False, max_request_id=100, lock=Lock())
         session.cluster.connection_factory.return_value = conn
-        pool = HostConnection(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         pool._keyspace = 'newks'
         conn.keyspace = 'foobarkeyspace'
         conn.set_keyspace_blocking.side_effect = \
@@ -520,7 +675,7 @@ def test_failed_keyspace_change_can_be_corrected_by_use(self):
         conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False,
                             is_closed=False, max_request_id=100, lock=Lock())
         session.cluster.connection_factory.return_value = conn
-        pool = HostConnection(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
 
         errors = []
         pool._set_keyspace_for_all_conns(
@@ -544,7 +699,7 @@ def test_failed_keyspace_send_still_completes_pool_callback(self):
         conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False,
                             is_closed=False, max_request_id=100, lock=Lock())
         session.cluster.connection_factory.return_value = conn
-        pool = HostConnection(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         error = ConnectionBusy('send blocked')
         conn.set_keyspace_async.side_effect = error
         callback = Mock()
@@ -562,7 +717,7 @@ def test_stale_shard_uses_aligned_connection(self):
                              is_closed=False, max_request_id=100, lock=Lock())
         stale.features.shard_id = 0
         session.cluster.connection_factory.return_value = stale
-        pool = HostConnection(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         aligned = HashableMock(spec=Connection, in_flight=0, is_defunct=False,
                                is_closed=False, max_request_id=100, lock=Lock())
         aligned.features.shard_id = 1
@@ -581,7 +736,7 @@ def test_published_connection_catches_newer_keyspace(self):
         conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False,
                             is_closed=False, max_request_id=100, lock=Lock())
         session.cluster.connection_factory.return_value = conn
-        pool = HostConnection(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         conn.keyspace = None
 
         def set_keyspace(keyspace):
@@ -601,7 +756,7 @@ def test_async_keyspace_reply_reapplies_current_target(self):
         conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False,
                             is_closed=False, max_request_id=100, lock=Lock())
         session.cluster.connection_factory.return_value = conn
-        pool = HostConnection(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         finished = []
 
         pool._set_keyspace_for_all_conns(
@@ -635,7 +790,7 @@ def test_failed_replacement_keyspace_sync_removes_connection(self):
                                 is_closed=False, max_request_id=100, lock=Lock())
         new_conn.features.shard_id = 0
         session.cluster.connection_factory.side_effect = [old_conn, new_conn]
-        pool = HostConnection(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
 
         def set_keyspace(keyspace):
             if keyspace == 'foobarkeyspace':
@@ -662,7 +817,7 @@ def test_failed_shard_keyspace_sync_allows_retry(self):
                                   is_closed=False, max_request_id=100, lock=Lock())
         shard_conn.features.shard_id = 1
         session.cluster.connection_factory.side_effect = [first_conn, shard_conn]
-        pool = HostConnection(host, HostDistance.LOCAL, session)
+        pool = self.make_pool(host, session)
         host.sharding_info = Mock(shards_count=2)
         pool._get_shard_aware_endpoint = Mock(return_value=None)
         pool._connecting.add(1)
diff --git a/tests/unit/test_lifecycle_coordination.py b/tests/unit/test_lifecycle_coordination.py
index 0fee02480a..f893632dce 100644
--- a/tests/unit/test_lifecycle_coordination.py
+++ b/tests/unit/test_lifecycle_coordination.py
@@ -1122,6 +1122,75 @@ def reply():
         pool.shutdown()
 
 
+def test_unpublished_failure_keeps_session_available_for_publication(topology):
+    cluster, _ = topology
+    host = cluster.metadata.get_host_by_host_id(HOST_ID_3)
+    session = new_session_with_pool_state()
+    session.cluster = cluster
+    session._profile_manager = cluster.profile_manager
+    session.keyspace = None
+    session.is_shutdown = False
+    session.shutdown = Mock()
+    connection = _PublicationConnection(host.endpoint, compression=False)
+    cluster.connection_factory = Mock(return_value=connection)
+    pool = HostConnection(host, HostDistance.LOCAL, session)
+    connection.is_defunct = True
+    host.signal_connection_failure = Mock()
+    work_lock = pool._lock
+    attempting = Event()
+    published = Event()
+    errors = []
+
+    class ObservedLock:
+        def __enter__(self):
+            attempting.set()
+            work_lock.acquire()
+
+        def __exit__(self, *args):
+            work_lock.release()
+
+    pool._lock = ObservedLock()
+
+    def fail():
+        try:
+            pool.return_connection(connection, stream_was_orphaned=True)
+        except BaseException as exc:
+            errors.append(exc)
+
+    def publish():
+        try:
+            with patch('cassandra.cluster.HostConnection', return_value=pool):
+                assert session.add_or_renew_pool(host, False).result(timeout=5)
+            published.set()
+        except BaseException as exc:
+            errors.append(exc)
+
+    failure = Thread(target=fail)
+    publication = Thread(target=publish)
+    try:
+        # A background shard may own the work lock while publication takes
+        # membership then Session. A stale failure must release Session before
+        # waiting for work-lock cleanup, or all three operations can deadlock.
+        work_lock.acquire()
+        try:
+            failure.start()
+            assert attempting.wait(5)
+            publication.start()
+            assert published.wait(2)
+        finally:
+            work_lock.release()
+    finally:
+        for worker in (failure, publication):
+            if worker.ident is not None:
+                worker.join(5)
+        pool.shutdown()
+    assert not failure.is_alive()
+    assert not publication.is_alive()
+    assert not errors
+    assert session._pools[host] is pool
+    host.signal_connection_failure.assert_not_called()
+
+
 @pytest.mark.parametrize('healthy', [False, True])
 def test_publication_progress_with_unsynchronized_background_shard(
         topology, healthy):

From f6ddcb1933c3ad3d5314a3c540c61b24185b5445 Mon Sep 17 00:00:00 2001
From: Dmitry Kropachev <dmitry.kropachev@gmail.com>
Date: Sat, 24 Oct 2026 23:16:25 -0400
Subject: [PATCH 2/7] cluster: exclude failing pool sockets from DOWN
 discounting

DOWN inspection runs before current connection-failure cleanup. Healthy
shard sockets in that same pool can therefore suppress recovery just
before the pool shuts down. Ignore the failure reservation's session
during health inspection while retaining discounting from other sessions
and the final ownership check at admission.

Exercise real pool state and DOWN admission for a failed shard alongside
an open sibling, with and without an independent healthy session.

Refs: #858
---
 cassandra/cluster.py                    |  6 +++
 tests/unit/test_host_connection_pool.py | 57 +++++++++++++++++++++++--
 2 files changed, 60 insertions(+), 3 deletions(-)

diff --git a/cassandra/cluster.py b/cassandra/cluster.py
index 1962f95a65..67d933bc88 100644
--- a/cassandra/cluster.py
+++ b/cassandra/cluster.py
@@ -2655,6 +2655,12 @@ def on_down(self, host, is_host_addition, expect_host_to_be_down=False,
         if (self._discount_down_events and
                 self.profile_manager.distance(host) != HostDistance.IGNORED):
             for session in tuple(self.sessions):
+                # This failure will shut down its pool, including any healthy
+                # sibling shards. Only independent pools can discount DOWN.
+                if (isinstance(_pool_failure,
+                               _lifecycle.ConnectionFailureReservation) and
+                        session is _pool_failure.session):
+                    continue
                 pool_state = session.get_pool_state().get(host)
                 if pool_state:
                     connected |= pool_state['open_count'] > 0
diff --git a/tests/unit/test_host_connection_pool.py b/tests/unit/test_host_connection_pool.py
index 10a520c405..03c245fda7 100644
--- a/tests/unit/test_host_connection_pool.py
+++ b/tests/unit/test_host_connection_pool.py
@@ -30,15 +30,17 @@
 import pytest
 
 from tests.unit.util import HashableMock
+from tests.unit.utils import new_session_with_pool_state
 
 LOGGER = logging.getLogger(__name__)
 
 
-def _make_socket_pool():
+def _make_socket_pool(session=None):
     host = Host('127.0.0.1', SimpleConvictionPolicy, host_id=uuid.uuid4())
     host.signal_connection_failure = Mock(return_value=False)
-    session = NonCallableMagicMock(
-        spec=Session, keyspace=None, _lock=RLock(), is_shutdown=False)
+    if session is None:
+        session = NonCallableMagicMock(
+            spec=Session, keyspace=None, _lock=RLock(), is_shutdown=False)
     session.cluster.is_shutdown = False
     session.cluster.shard_aware_options.disable = True
 
@@ -209,6 +211,55 @@ def distance(current):
         cluster.shutdown()
 
 
+@pytest.mark.parametrize('independent_pool', [False, True])
+def test_failed_pool_cannot_discount_its_own_down(independent_pool):
+    session = new_session_with_pool_state()
+    session.cluster = Mock()
+    session.keyspace = None
+    session.is_shutdown = False
+    host, session, pool, original, sibling = _make_socket_pool(session)
+    sibling.features = ProtocolFeatures(shard_id=1)
+    pool._connections[1] = sibling
+
+    other_session = new_session_with_pool_state()
+    other_session.cluster = session.cluster
+    other_session.keyspace = None
+    other_session.is_shutdown = False
+    healthy = HashableMock(
+        spec=Connection, in_flight=0, is_closed=False, is_defunct=False,
+        orphaned_request_ids=set(), features=ProtocolFeatures(shard_id=0))
+    session.cluster.connection_factory.side_effect = [healthy]
+    other_pool = HostConnection(host, HostDistance.LOCAL, other_session)
+    other_session._pools[host] = other_pool
+
+    cluster = Cluster(contact_points=[])
+    session.cluster = other_session.cluster = cluster
+    cluster.sessions = ((session, other_session) if independent_pool
+                        else (session,))
+    cluster.profile_manager.distance = Mock(return_value=HostDistance.LOCAL)
+    cluster.on_down_potentially_blocking = Mock(return_value=Future())
+    host.set_up()
+    host.signal_connection_failure.return_value = True
+    original.in_flight = 1
+    original.is_defunct = True
+    original.last_error = ConnectionException('one shard failed', host.endpoint)
+    try:
+        assert cluster._discount_down_events
+        assert session.get_pool_state()[host]['open_count'] == 1
+        pool.return_connection(original)
+        assert original.in_flight == 0
+        assert pool.is_shutdown
+        assert host.is_up is independent_pool
+        assert (cluster.on_down_potentially_blocking.call_count ==
+                (0 if independent_pool else 1))
+        assert not other_pool.is_shutdown
+        healthy.close.assert_not_called()
+    finally:
+        pool.shutdown()
+        other_pool.shutdown()
+        cluster.shutdown()
+
+
 def test_queued_replacement_cannot_remove_new_socket(socket_pool):
     host, session, pool, original, replacement = socket_pool
     pool.borrow_connection(timeout=0)

From 75a0d2b9cbbdfc27a972ee6cd4a73241a0cffaf8 Mon Sep 17 00:00:00 2001
From: Dmitry Kropachev <dmitry.kropachev@gmail.com>
Date: Sat, 24 Oct 2026 23:53:29 -0400
Subject: [PATCH 3/7] lifecycle: release ownership guards while waiting for
 pool work

A shard connection can hold its pool work lock while waiting for a USE
response. Waiting for that lock with Session, Host, or lifecycle guards
held can block a failure callback for another host on the same reactor.

Release those guards before waiting, then recheck ownership and retry the
complete DOWN admission transaction. Keep state changes and FIFO admission
atomic, and test both continued recovery and superseded ownership.
---
 cassandra/_lifecycle.py                   | 118 +++++++++-------
 tests/unit/test_lifecycle_coordination.py | 162 +++++++++++++++++++++-
 2 files changed, 231 insertions(+), 49 deletions(-)

diff --git a/cassandra/_lifecycle.py b/cassandra/_lifecycle.py
index a9dab7b932..869bdb4c04 100644
--- a/cassandra/_lifecycle.py
+++ b/cassandra/_lifecycle.py
@@ -39,6 +39,10 @@ class EndpointMoveRejected(ValueError):
     """A batch could not publish because its metadata snapshot went stale."""
 
 
+class ConnectionFailureLockContended(Exception):
+    """Retry admission after releasing the enclosing lifecycle transaction."""
+
+
 class PoolFailureReservation:
     """Fence an admitted pool failure without locking across policy callbacks."""
 
@@ -60,7 +64,7 @@ def is_current(self):
             return self.owns_locked()
 
     @contextmanager
-    def locked(self):
+    def locked(self, wait=True):
         with self.session._lock:
             with self.host.lock:
                 yield
@@ -77,20 +81,34 @@ def __init__(self, pool, connection, replacing=False):
         self.replacing = replacing
 
     @contextmanager
-    def locked(self):
+    def locked(self, wait=True):
         if self.session is None:
             yield
             return
-        with self.session._lock:
-            with self.host.lock:
-                # Unpublished candidates can hold their work lock while
-                # publication waits for Session under the membership lock.
-                # Only a current pool needs its shard ownership inspected.
-                if self.session._pools.get(self.host) is not self.pool:
-                    yield
-                    return
-                with self.pool._lock:
-                    yield
+        while True:
+            with self.session._lock:
+                with self.host.lock:
+                    # Unpublished candidates can hold their work lock while
+                    # publication waits for Session under the membership lock.
+                    # Only a current pool needs its shard ownership inspected.
+                    if self.session._pools.get(self.host) is not self.pool:
+                        yield
+                        return
+                    if self.pool._lock.acquire(False):
+                        try:
+                            yield
+                        finally:
+                            self.pool._lock.release()
+                        return
+            if not wait:
+                raise ConnectionFailureLockContended()
+            self.wait()
+
+    def wait(self):
+        # Never retain work while reacquiring Session/Host. DOWN callers also
+        # leave their lifecycle transaction before waiting and retry admission.
+        with self.pool._lock:
+            pass
 
     def owns_locked(self):
         if (self.session is None or
@@ -247,7 +265,7 @@ def observe_health(self, host):
     def begin_down(self, host, health_generation, connected=False,
                    expect_down=False, pool_failure=None):
         """Return (generation, retry_only), or None if DOWN is discounted."""
-        failure_lock = (pool_failure.locked()
+        failure_lock = (pool_failure.locked(wait=False)
                         if pool_failure is not None else host.lock)
         with failure_lock:
             if (self.closed or host._is_removed or connected or
@@ -273,41 +291,45 @@ def dispatch_down(self, host, health_generation, is_host_addition,
                       pool_failure=None):
         # Publish DOWN state and FIFO admission together. Do not drain on
         # this caller; the executor (or another drainer) owns delivery.
-        with self.transaction(drain=False):
-            admission = self.begin_down(
-                host, health_generation, connected=connected,
-                expect_down=expect_host_to_be_down,
-                pool_failure=pool_failure)
-            if admission is None:
-                return False
-            down_event_generation, restart_reconnector = admission
-
-            if restart_reconnector:
-                if on_add_reconnection is None:
-                    future = self._cluster._restart_reconnector(
-                        host, is_host_addition, down_event_generation)
-                else:
-                    future = self._cluster._restart_reconnector(
-                        host, is_host_addition, down_event_generation,
-                        on_add_reconnection=on_add_reconnection)
-                if future is None:
-                    self.finish_down(host, down_event_generation)
-                # 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)
-
-            if on_add_reconnection is None:
-                future = self._cluster.on_down_potentially_blocking(
-                    host, is_host_addition, down_event_generation)
-            else:
-                future = self._cluster.on_down_potentially_blocking(
-                    host, is_host_addition, down_event_generation,
-                    on_add_reconnection=on_add_reconnection)
-            if future is None:
-                self.finish_down(host, down_event_generation)
-            return future is not None
+        while True:
+            try:
+                with self.transaction(drain=False):
+                    admission = self.begin_down(
+                        host, health_generation, connected=connected,
+                        expect_down=expect_host_to_be_down,
+                        pool_failure=pool_failure)
+                    if admission is None:
+                        return False
+                    down_event_generation, restart_reconnector = admission
+
+                    if restart_reconnector:
+                        if on_add_reconnection is None:
+                            future = self._cluster._restart_reconnector(
+                                host, is_host_addition, down_event_generation)
+                        else:
+                            future = self._cluster._restart_reconnector(
+                                host, is_host_addition, down_event_generation,
+                                on_add_reconnection=on_add_reconnection)
+                        if future is None:
+                            self.finish_down(host, down_event_generation)
+                        # Restarting the host reconnector does not dispatch
+                        # DOWN callbacks that reconnect the control connection.
+                        return False
+
+                    log.warning("Host %s has been marked down", host)
+
+                    if on_add_reconnection is None:
+                        future = self._cluster.on_down_potentially_blocking(
+                            host, is_host_addition, down_event_generation)
+                    else:
+                        future = self._cluster.on_down_potentially_blocking(
+                            host, is_host_addition, down_event_generation,
+                            on_add_reconnection=on_add_reconnection)
+                    if future is None:
+                        self.finish_down(host, down_event_generation)
+                    return future is not None
+            except ConnectionFailureLockContended:
+                pool_failure.wait()
 
     @staticmethod
     def finish_down(host, generation):
diff --git a/tests/unit/test_lifecycle_coordination.py b/tests/unit/test_lifecycle_coordination.py
index f893632dce..c9942bded8 100644
--- a/tests/unit/test_lifecycle_coordination.py
+++ b/tests/unit/test_lifecycle_coordination.py
@@ -13,7 +13,7 @@
 # limitations under the License.
 
 from concurrent.futures import Future
-from threading import Event, Lock, Thread
+from threading import Event, Lock, Thread, get_ident
 from unittest.mock import Mock, patch
 
 import pytest
@@ -1191,6 +1191,166 @@ def publish():
     host.signal_connection_failure.assert_not_called()
 
 
+@pytest.mark.parametrize('phase, supersede', [
+    ('return', None), ('down', None), ('return', 'pool'),
+    ('down', 'pool'), ('down', 'health'),
+])
+def test_current_failure_contention_allows_other_host_response(
+        topology, phase, supersede):
+    from cassandra.protocol import ResultMessage, RESULT_KIND_SET_KEYSPACE
+    from cassandra.shard_info import _ShardingInfo
+
+    cluster, _ = topology
+    hosts = [cluster.metadata.get_host_by_host_id(host_id)
+             for host_id in (HOST_ID_1, HOST_ID_2)]
+    session = new_session_with_pool_state()
+    session.cluster = cluster
+    session._profile_manager = cluster.profile_manager
+    session.keyspace = 'ks'
+    session.is_shutdown = False
+    session.shutdown = Mock()
+    cluster.shard_aware_options.disable_shardaware_port = True
+    connections = [_PublicationConnection(host.endpoint, compression=False)
+                   for host in hosts]
+    for connection in connections:
+        connection.features = ProtocolFeatures(shard_id=0)
+        connection.keyspace = 'ks'
+        connection.is_defunct = True
+        connection.in_flight = 1
+    candidate = _PublicationConnection(hosts[0].endpoint, compression=False)
+    candidate.features = ProtocolFeatures(shard_id=1)
+    cluster.connection_factory = Mock(side_effect=connections + [candidate])
+    pools = [HostConnection(host, HostDistance.LOCAL, session) for host in hosts]
+    session._pools.update(zip(hosts, pools))
+    hosts[0].sharding_info = _ShardingInfo(0, 2, '', '', 0, None, None)
+    successor = (HostConnection(hosts[0], HostDistance.IGNORED, session)
+                 if supersede == 'pool' else None)
+    for host in hosts:
+        host.signal_connection_failure = Mock(return_value=True)
+
+    io_waiting = Event()
+    at_down = Event()
+    contended = Event()
+    reactor_done = Event()
+    response_lock = Lock()
+    responses = []
+    delivered = []
+    delivery_futures = []
+    errors = []
+
+    def send(message, request_id, cb, **kwargs):
+        with response_lock:
+            responses.append(cb)
+        io_waiting.set()
+        return 0
+
+    candidate.send_msg = send
+
+    def respond():
+        with response_lock:
+            callbacks = list(responses)
+            responses.clear()
+        for callback in callbacks:
+            callback(ResultMessage(RESULT_KIND_SET_KEYSPACE))
+
+    def admit(host, is_addition, generation):
+        future = cluster._lifecycle.submit_async(
+            cluster.executor, delivered.append, host)
+        delivery_futures.append(future)
+        return future
+
+    cluster.on_down_potentially_blocking = Mock(side_effect=admit)
+    begin_down = cluster._lifecycle.begin_down
+
+    def begin(host, *args, **kwargs):
+        if phase == 'down' and host is hosts[0]:
+            at_down.set()
+            assert io_waiting.wait(5)
+        return begin_down(host, *args, **kwargs)
+
+    cluster._lifecycle.begin_down = begin
+    work_lock = pools[0]._lock
+
+    class ObservedLock:
+        def acquire(self, blocking=True):
+            if failure.ident == get_ident() and io_waiting.is_set():
+                contended.set()
+            return work_lock.acquire(blocking)
+
+        def release(self):
+            work_lock.release()
+
+        def __enter__(self):
+            self.acquire()
+
+        def __exit__(self, *args):
+            self.release()
+
+    pools[0]._lock = ObservedLock()
+
+    def run(fn):
+        try:
+            fn()
+        except BaseException as exc:
+            errors.append(exc)
+
+    def open_shard():
+        if phase == 'down':
+            assert at_down.wait(5)
+        pools[0]._open_connection_to_missing_shard(1)
+
+    def process_responses():
+        # One reactor must finish the other host's failure callback before
+        # it can deliver the USE response that releases the pool work lock.
+        if supersede == 'pool':
+            with session._lock, hosts[0].lock:
+                session._pools[hosts[0]] = successor
+        elif supersede == 'health':
+            assert cluster._lifecycle.mark_ready(hosts[0])
+        pools[1].return_connection(connections[1])
+        respond()
+        reactor_done.set()
+
+    failure = Thread(target=run, args=(lambda: pools[0].return_connection(
+        connections[0]),))
+    worker = Thread(target=run, args=(open_shard,))
+    reactor = Thread(target=run, args=(process_responses,))
+    try:
+        worker.start()
+        if phase == 'return':
+            assert io_waiting.wait(5)
+        failure.start()
+        assert io_waiting.wait(5)
+        assert contended.wait(5)
+        reactor.start()
+        assert reactor_done.wait(2), 'cross-host failure blocked reactor progress'
+    finally:
+        # Release controlled I/O even for the unfixed production negative
+        # control, so assertion failures cannot leave blocked driver threads.
+        respond()
+        for thread in (worker, failure, reactor):
+            if thread.ident is not None:
+                thread.join(5)
+        for pool in pools:
+            pool.shutdown()
+    assert all(not thread.is_alive() for thread in (worker, failure, reactor))
+    assert not errors
+    assert all(connection.in_flight == 0 for connection in connections)
+    for future in delivery_futures:
+        future.result(timeout=5)
+    assert delivered == ([hosts[1]] if supersede else [hosts[1], hosts[0]])
+    assert hosts[0].is_up is bool(supersede)
+    assert not hosts[1].is_up
+    hosts[1].signal_connection_failure.assert_called_once()
+    assert hosts[0].signal_connection_failure.call_count == int(
+        phase == 'down' or supersede != 'pool')
+    if successor is not None:
+        assert session._pools[hosts[0]] is successor
+        assert not successor.is_shutdown
+        successor.shutdown()
+    assert candidate.keyspace == 'ks'
+
+
 @pytest.mark.parametrize('healthy', [False, True])
 def test_publication_progress_with_unsynchronized_background_shard(
         topology, healthy):

From f99ba3c3fe489a58320714030f1e424ee8937bd1 Mon Sep 17 00:00:00 2001
From: Dmitry Kropachev <dmitry.kropachev@gmail.com>
Date: Sat, 24 Oct 2026 23:54:07 -0400
Subject: [PATCH 4/7] pool: retain the retired socket identity when retrying
 replacement

A regular-port candidate can land on a different shard. If keyspace sync
fails, retrying with that candidate loses the original shard reservation
and leaves the pool without a usable connection.

Keep the candidate separate from the retired socket so cleanup closes the
failed candidate and the queued retry retains its original owner. Extend
the keyspace-sync regression to run the queued retry and borrow its usable
replacement.
---
 cassandra/pool.py                       |  9 ++---
 tests/unit/test_host_connection_pool.py | 48 ++++++++++++++++---------
 2 files changed, 36 insertions(+), 21 deletions(-)

diff --git a/cassandra/pool.py b/cassandra/pool.py
index 6b78126581..be1ce34fc4 100644
--- a/cassandra/pool.py
+++ b/cassandra/pool.py
@@ -761,12 +761,13 @@ def _replace(self, connection):
                     self._connecting.add(connection.features.shard_id)
                     self._session.submit(self._open_connection_to_missing_shard, connection.features.shard_id)
                 else:
-                    connection = self._session.cluster.connection_factory(self.endpoint,
-                                                                          on_orphaned_stream_released=self.on_orphaned_stream_released)
+                    new_connection = self._session.cluster.connection_factory(
+                        self.endpoint,
+                        on_orphaned_stream_released=self.on_orphaned_stream_released)
                     try:
-                        self._sync_keyspace(connection)
+                        self._sync_keyspace(new_connection)
                     except Exception:
-                        failed_connection = connection
+                        failed_connection = new_connection
                         raise
             except Exception:
                 log.warning("Failed reconnecting %s. Retrying." % (self.endpoint,))
diff --git a/tests/unit/test_host_connection_pool.py b/tests/unit/test_host_connection_pool.py
index 03c245fda7..b31e960432 100644
--- a/tests/unit/test_host_connection_pool.py
+++ b/tests/unit/test_host_connection_pool.py
@@ -830,18 +830,17 @@ def test_async_keyspace_reply_reapplies_current_target(self):
         assert conn.keyspace == pool._keyspace
 
     def test_failed_replacement_keyspace_sync_removes_connection(self):
-        host = Mock(spec=Host, address='ip1', sharding_info=None)
-        session = self.make_session()
-        session.cluster.shard_aware_options.disable = True
-        old_conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False,
-                                is_closed=False, max_request_id=100, lock=Lock())
-        old_conn.features.shard_id = 0
-        old_conn.features.sharding_info = None
-        new_conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False,
-                                is_closed=False, max_request_id=100, lock=Lock())
-        new_conn.features.shard_id = 0
-        session.cluster.connection_factory.side_effect = [old_conn, new_conn]
-        pool = self.make_pool(host, session)
+        host, session, pool, old_conn, new_conn = _make_socket_pool()
+        pool._keyspace = 'foobarkeyspace'
+        # Regular-port replacements can land on a different shard.
+        new_conn.features = ProtocolFeatures(shard_id=1)
+        healthy = HashableMock(
+            spec=Connection, in_flight=0, is_defunct=False, is_closed=False,
+            max_request_id=100, lock=Lock(), keyspace=None,
+            features=ProtocolFeatures(shard_id=0))
+        healthy.set_keyspace_blocking.side_effect = \
+            lambda keyspace: setattr(healthy, 'keyspace', keyspace)
+        session.cluster.connection_factory.side_effect = [new_conn, healthy]
 
         def set_keyspace(keyspace):
             if keyspace == 'foobarkeyspace':
@@ -851,11 +850,26 @@ def set_keyspace(keyspace):
                 raise ConnectionException('sync failed', host.endpoint)
 
         new_conn.set_keyspace_blocking.side_effect = set_keyspace
-        pool._replace(old_conn)
-
-        assert 0 not in pool._connections
-        new_conn.close.assert_called_once()
-        session.submit.assert_called_once_with(pool._replace, new_conn)
+        old_conn.is_defunct = True
+        old_conn.in_flight = 1
+        try:
+            pool.return_connection(old_conn)
+            replace, retired = session.submit.call_args.args
+            replace(retired)
+
+            assert not pool._connections
+            new_conn.close.assert_called_once()
+            replace, retired = session.submit.call_args.args
+            replace(retired)
+
+            borrowed, _ = pool.borrow_connection(timeout=0)
+            assert borrowed is healthy
+            assert healthy.keyspace == pool._keyspace == 'newks'
+            assert not pool._replacement_connections
+            healthy.close.assert_not_called()
+            pool.return_connection(borrowed)
+        finally:
+            pool.shutdown()
 
     def test_failed_shard_keyspace_sync_allows_retry(self):
         host = Mock(spec=Host, address='ip1', sharding_info=None)

From aeeb78939104e9aa8369b2f11f9cbbd29a35c0f1 Mon Sep 17 00:00:00 2001
From: Dmitry Kropachev <dmitry.kropachev@gmail.com>
Date: Sun, 25 Oct 2026 01:36:59 -0400
Subject: [PATCH 5/7] pool: retire failed pools before inspecting host health

Retire the owning pool before DOWN health inspection so simultaneous
failures cannot each count the other's remaining shard as healthy. Keep
the retirement receipt on the failure reservation and recheck Session,
Host, and endpoint ownership after unlocked cleanup.

Return already-signaled no-op callbacks before waiting for pool work,
with the guarded duplicate check retained for first-callback races.
Retired reservations need only Session and Host guards, removing the
coordinator contention retry while preserving active-pool progress.

Add controlled USE, concurrent-session, and retired-admission regressions.
---
 cassandra/_lifecycle.py                   | 119 ++++-----
 cassandra/pool.py                         |  25 +-
 tests/unit/test_host_connection_pool.py   | 151 ++++++++++-
 tests/unit/test_lifecycle_coordination.py | 293 +++++++++++++++++++---
 4 files changed, 479 insertions(+), 109 deletions(-)

diff --git a/cassandra/_lifecycle.py b/cassandra/_lifecycle.py
index 869bdb4c04..e96884a900 100644
--- a/cassandra/_lifecycle.py
+++ b/cassandra/_lifecycle.py
@@ -39,10 +39,6 @@ class EndpointMoveRejected(ValueError):
     """A batch could not publish because its metadata snapshot went stale."""
 
 
-class ConnectionFailureLockContended(Exception):
-    """Retry admission after releasing the enclosing lifecycle transaction."""
-
-
 class PoolFailureReservation:
     """Fence an admitted pool failure without locking across policy callbacks."""
 
@@ -64,7 +60,7 @@ def is_current(self):
             return self.owns_locked()
 
     @contextmanager
-    def locked(self, wait=True):
+    def locked(self):
         with self.session._lock:
             with self.host.lock:
                 yield
@@ -79,9 +75,10 @@ def __init__(self, pool, connection, replacing=False):
         self.session = pool._session_ref()
         self.host = pool.host
         self.replacing = replacing
+        self.retired = False
 
     @contextmanager
-    def locked(self, wait=True):
+    def locked(self):
         if self.session is None:
             yield
             return
@@ -90,8 +87,10 @@ def locked(self, wait=True):
                 with self.host.lock:
                     # Unpublished candidates can hold their work lock while
                     # publication waits for Session under the membership lock.
-                    # Only a current pool needs its shard ownership inspected.
-                    if self.session._pools.get(self.host) is not self.pool:
+                    # A completed retirement no longer needs shard ownership
+                    # inspected; its receipt retains only Session/Host guards.
+                    if (self.retired or
+                            self.session._pools.get(self.host) is not self.pool):
                         yield
                         return
                     if self.pool._lock.acquire(False):
@@ -100,30 +99,26 @@ def locked(self, wait=True):
                         finally:
                             self.pool._lock.release()
                         return
-            if not wait:
-                raise ConnectionFailureLockContended()
-            self.wait()
-
-    def wait(self):
-        # Never retain work while reacquiring Session/Host. DOWN callers also
-        # leave their lifecycle transaction before waiting and retry admission.
-        with self.pool._lock:
-            pass
+            # Never retain work while reacquiring Session/Host. Active socket
+            # failures may await I/O under work before they can retire a pool.
+            with self.pool._lock:
+                pass
 
     def owns_locked(self):
         if (self.session is None or
                 self.session._pools.get(self.host) is not self.pool):
             return False
-        current = self.pool._connections.get(self.connection.features.shard_id)
-        owns_slot = (current is self.connection or
-                     (self.replacing and current is None and
-                      self.pool._replacement_connections.get(
-                          self.connection.features.shard_id) is self.connection))
+        owns_slot = self.retired
+        if not self.retired and not self.pool.is_shutdown:
+            current = self.pool._connections.get(self.connection.features.shard_id)
+            owns_slot = (current is self.connection or
+                         (self.replacing and current is None and
+                          self.pool._replacement_connections.get(
+                              self.connection.features.shard_id) is self.connection))
         return (not self.session.is_shutdown and
                 not self.session.cluster.is_shutdown and
                 not self.host._is_removed and
-                self.host.endpoint == self.pool.endpoint and
-                not self.pool.is_shutdown and owns_slot)
+                self.host.endpoint == self.pool.endpoint and owns_slot)
 
     def is_current(self):
         with self.locked():
@@ -265,7 +260,7 @@ def observe_health(self, host):
     def begin_down(self, host, health_generation, connected=False,
                    expect_down=False, pool_failure=None):
         """Return (generation, retry_only), or None if DOWN is discounted."""
-        failure_lock = (pool_failure.locked(wait=False)
+        failure_lock = (pool_failure.locked()
                         if pool_failure is not None else host.lock)
         with failure_lock:
             if (self.closed or host._is_removed or connected or
@@ -291,45 +286,41 @@ def dispatch_down(self, host, health_generation, is_host_addition,
                       pool_failure=None):
         # Publish DOWN state and FIFO admission together. Do not drain on
         # this caller; the executor (or another drainer) owns delivery.
-        while True:
-            try:
-                with self.transaction(drain=False):
-                    admission = self.begin_down(
-                        host, health_generation, connected=connected,
-                        expect_down=expect_host_to_be_down,
-                        pool_failure=pool_failure)
-                    if admission is None:
-                        return False
-                    down_event_generation, restart_reconnector = admission
-
-                    if restart_reconnector:
-                        if on_add_reconnection is None:
-                            future = self._cluster._restart_reconnector(
-                                host, is_host_addition, down_event_generation)
-                        else:
-                            future = self._cluster._restart_reconnector(
-                                host, is_host_addition, down_event_generation,
-                                on_add_reconnection=on_add_reconnection)
-                        if future is None:
-                            self.finish_down(host, down_event_generation)
-                        # Restarting the host reconnector does not dispatch
-                        # DOWN callbacks that reconnect the control connection.
-                        return False
-
-                    log.warning("Host %s has been marked down", host)
-
-                    if on_add_reconnection is None:
-                        future = self._cluster.on_down_potentially_blocking(
-                            host, is_host_addition, down_event_generation)
-                    else:
-                        future = self._cluster.on_down_potentially_blocking(
-                            host, is_host_addition, down_event_generation,
-                            on_add_reconnection=on_add_reconnection)
-                    if future is None:
-                        self.finish_down(host, down_event_generation)
-                    return future is not None
-            except ConnectionFailureLockContended:
-                pool_failure.wait()
+        with self.transaction(drain=False):
+            admission = self.begin_down(
+                host, health_generation, connected=connected,
+                expect_down=expect_host_to_be_down,
+                pool_failure=pool_failure)
+            if admission is None:
+                return False
+            down_event_generation, restart_reconnector = admission
+
+            if restart_reconnector:
+                if on_add_reconnection is None:
+                    future = self._cluster._restart_reconnector(
+                        host, is_host_addition, down_event_generation)
+                else:
+                    future = self._cluster._restart_reconnector(
+                        host, is_host_addition, down_event_generation,
+                        on_add_reconnection=on_add_reconnection)
+                if future is None:
+                    self.finish_down(host, down_event_generation)
+                # Restarting the host reconnector does not dispatch
+                # DOWN callbacks that reconnect the control connection.
+                return False
+
+            log.warning("Host %s has been marked down", host)
+
+            if on_add_reconnection is None:
+                future = self._cluster.on_down_potentially_blocking(
+                    host, is_host_addition, down_event_generation)
+            else:
+                future = self._cluster.on_down_potentially_blocking(
+                    host, is_host_addition, down_event_generation,
+                    on_add_reconnection=on_add_reconnection)
+            if future is None:
+                self.finish_down(host, down_event_generation)
+            return future is not None
 
     @staticmethod
     def finish_down(host, generation):
diff --git a/cassandra/pool.py b/cassandra/pool.py
index be1ce34fc4..187c55e033 100644
--- a/cassandra/pool.py
+++ b/cassandra/pool.py
@@ -655,6 +655,11 @@ def return_connection(self, connection, stream_was_orphaned=False,
                 self._stream_available_condition.notify()
 
         if connection.is_defunct or connection.is_closed:
+            if connection.signaled_error and not replace_only and \
+                    (not self.shutdown_on_error or
+                     (isinstance(connection.last_error, ConnectionSendError) and
+                      not heartbeat_failed)):
+                return
             reservation = _lifecycle.ConnectionFailureReservation(self, connection)
             with reservation.locked():
                 current = reservation.owns_locked()
@@ -684,16 +689,14 @@ def return_connection(self, connection, stream_was_orphaned=False,
                 is_down = True
 
             if is_down:
-                # Policy and health inspection can replace the pool or socket.
-                # Carry ownership through DOWN admission and pool shutdown.
+                # Retire before inspecting other sessions: concurrent failures
+                # must not discount each other's doomed sibling connections.
+                self.shutdown(_failure=reservation)
                 if reservation.is_current():
-                    try:
-                        self._session.cluster.on_down(
-                            self.host, is_host_addition=False,
-                            _pool_failure=reservation)
-                    finally:
-                        self.shutdown(_failure=reservation)
-                else:
+                    self._session.cluster.on_down(
+                        self.host, is_host_addition=False,
+                        _pool_failure=reservation)
+                elif not reservation.retired:
                     with self._lock:
                         self._trash.discard(connection)
                     connection.close()
@@ -789,6 +792,10 @@ def shutdown(self, _failure=None):
             with self._membership_lock:
                 if self.is_shutdown:
                     return
+                if _failure is not None:
+                    # Only this reservation may retain failure authority after
+                    # its own guarded retirement clears the shard slots.
+                    _failure.retired = True
                 self.is_shutdown = True
                 connections_to_close = self._connections.copy()
                 self._connections.clear()
diff --git a/tests/unit/test_host_connection_pool.py b/tests/unit/test_host_connection_pool.py
index b31e960432..62e32028a0 100644
--- a/tests/unit/test_host_connection_pool.py
+++ b/tests/unit/test_host_connection_pool.py
@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 from concurrent.futures import Future, ThreadPoolExecutor
+from contextlib import contextmanager
 import logging
 import uuid
 from cassandra.protocol_features import ProtocolFeatures
@@ -19,10 +20,11 @@
 from cassandra.shard_info import _ShardingInfo
 
 import unittest
-from threading import Thread, Event, Lock, RLock
+from threading import Barrier, Thread, Event, Lock, RLock, get_ident
 from unittest.mock import Mock, NonCallableMagicMock, MagicMock, patch
 
 from cassandra.cluster import Cluster, Session, ShardAwareOptions
+from cassandra._lifecycle import ConnectionFailureReservation
 from cassandra.connection import Connection, ConnectionBusy, ConnectionException, ConnectionSendError, DefaultEndPoint
 from cassandra.pool import HostConnection
 from cassandra.pool import Host, NoConnectionsAvailable
@@ -176,7 +178,7 @@ def convict(error):
         replacement.close.assert_not_called()
 
 
-@pytest.mark.parametrize('change', ['socket', 'pool'])
+@pytest.mark.parametrize('change', ['pool', 'endpoint', 'health'])
 def test_down_admission_rechecks_failed_socket(socket_pool, change):
     host, session, pool, original, replacement = socket_pool
     cluster = Cluster(contact_points=[])
@@ -192,18 +194,22 @@ def test_down_admission_rechecks_failed_socket(socket_pool, change):
 
     def distance(current):
         with session._lock, host.lock, pool._lock:
-            if change == 'socket':
-                pool._connections[0] = replacement
-            else:
+            if change == 'pool':
                 session._pools[host] = Mock(
                     host=host, endpoint=host.endpoint, is_shutdown=False)
+            elif change == 'endpoint':
+                host.endpoint = DefaultEndPoint('127.0.0.2')
+            else:
+                host._health_generation += 1
         return HostDistance.IGNORED
 
     cluster.profile_manager.distance.side_effect = distance
     try:
         pool.return_connection(original)
         assert host.is_up
-        assert not session._pools[host].is_shutdown
+        assert pool.is_shutdown
+        if change == 'pool':
+            assert not session._pools[host].is_shutdown
         replacement.close.assert_not_called()
         cluster.on_down_potentially_blocking.assert_not_called()
     finally:
@@ -211,6 +217,139 @@ def distance(current):
         cluster.shutdown()
 
 
+@pytest.mark.parametrize('change', ['pool', 'endpoint', 'host', 'session', 'cluster'])
+def test_retirement_cleanup_cannot_down_superseding_owner(socket_pool, change):
+    host, session, pool, original, replacement = socket_pool
+    host.signal_connection_failure.return_value = True
+    pool.borrow_connection(timeout=0)
+    original.is_defunct = True
+    successor = Mock(is_shutdown=False)
+
+    def supersede():
+        # Socket cleanup is outside ownership guards, so it can synchronously
+        # change topology or publish a successor before DOWN inspection starts.
+        with session._lock, host.lock:
+            assert pool._lock.acquire(False), 'cleanup retained the pool work lock'
+            try:
+                if change == 'pool':
+                    session._pools[host] = successor
+                elif change == 'endpoint':
+                    host.endpoint = DefaultEndPoint('127.0.0.2')
+                elif change == 'host':
+                    host._is_removed = True
+                elif change == 'session':
+                    session.is_shutdown = True
+                else:
+                    session.cluster.is_shutdown = True
+            finally:
+                pool._lock.release()
+
+    original.close.side_effect = supersede
+    pool.return_connection(original)
+    assert pool.is_shutdown
+    assert original.in_flight == 0
+    original.close.assert_called_once()
+    session.cluster.on_down.assert_not_called()
+    replacement.close.assert_not_called()
+    successor.shutdown.assert_not_called()
+
+
+@pytest.mark.parametrize('retirement', ['owned', 'generic', 'already_shutdown', 'stale', 'other_failure'])
+def test_only_own_guarded_retirement_can_admit_down(socket_pool, retirement):
+    host, session, pool, original, replacement = socket_pool
+    cluster = Cluster(contact_points=[])
+    session.cluster = cluster
+    cluster.sessions = (session,)
+    cluster.profile_manager.distance = Mock(return_value=HostDistance.IGNORED)
+    cluster.on_down_potentially_blocking = Mock(return_value=Future())
+    host.set_up()
+    failure = ConnectionFailureReservation(pool, original)
+    try:
+        if retirement == 'generic':
+            pool.shutdown()
+        elif retirement == 'already_shutdown':
+            pool.shutdown()
+            pool.shutdown(_failure=failure)
+        elif retirement == 'stale':
+            pool._connections[0] = replacement
+            pool.shutdown(_failure=failure)
+        elif retirement == 'other_failure':
+            pool.shutdown(_failure=ConnectionFailureReservation(pool, original))
+            pool.shutdown(_failure=failure)
+        else:
+            pool.shutdown(_failure=failure)
+
+        admitted = cluster.on_down(host, False, _pool_failure=failure)
+        assert admitted is (retirement == 'owned')
+        assert host.is_up is (retirement != 'owned')
+        assert cluster.on_down_potentially_blocking.call_count == int(retirement == 'owned')
+    finally:
+        pool.shutdown()
+        cluster.shutdown()
+
+
+def test_racing_first_callbacks_preserve_convicting_owner(socket_pool):
+    host, session, pool, original, _ = socket_pool
+    original.in_flight = 2
+    original.is_defunct = True
+    initial_callbacks = Barrier(2)
+    policy_entered, release_policy, second_done = Event(), Event(), Event()
+    errors = []
+    locked = ConnectionFailureReservation.locked
+
+    @contextmanager
+    def enter(reservation, *args, **kwargs):
+        if not getattr(reservation, '_test_entered', False):
+            reservation._test_entered = True
+            # Both callbacks have passed the unsignaled fast check before
+            # the first callback publishes its conviction-in-progress.
+            initial_callbacks.wait(timeout=5)
+            if get_ident() == second.ident:
+                assert policy_entered.wait(5)
+        with locked(reservation, *args, **kwargs):
+            yield
+
+    def convict(error):
+        policy_entered.set()
+        assert release_policy.wait(5)
+        return True
+
+    def fail(done=None):
+        try:
+            pool.return_connection(original)
+        except BaseException as exc:
+            errors.append(exc)
+        finally:
+            if done is not None:
+                done.set()
+
+    host.signal_connection_failure.side_effect = convict
+    first = Thread(target=fail)
+    second = Thread(target=fail, args=(second_done,))
+    try:
+        with patch.object(ConnectionFailureReservation, 'locked', enter):
+            first.start()
+            second.start()
+            assert second_done.wait(2)
+            assert pool._connections[0] is original
+            release_policy.set()
+            for worker in (first, second):
+                worker.join(5)
+    finally:
+        release_policy.set()
+        initial_callbacks.abort()
+        for worker in (first, second):
+            if worker.ident is not None:
+                worker.join(5)
+        pool.shutdown()
+    assert not errors
+    assert all(not worker.is_alive() for worker in (first, second))
+    assert original.in_flight == 0
+    host.signal_connection_failure.assert_called_once()
+    session.cluster.on_down.assert_called_once()
+    session.submit.assert_not_called()
+
+
 @pytest.mark.parametrize('independent_pool', [False, True])
 def test_failed_pool_cannot_discount_its_own_down(independent_pool):
     session = new_session_with_pool_state()
diff --git a/tests/unit/test_lifecycle_coordination.py b/tests/unit/test_lifecycle_coordination.py
index c9942bded8..447d218b7b 100644
--- a/tests/unit/test_lifecycle_coordination.py
+++ b/tests/unit/test_lifecycle_coordination.py
@@ -13,13 +13,14 @@
 # limitations under the License.
 
 from concurrent.futures import Future
-from threading import Event, Lock, Thread, get_ident
+from threading import Barrier, Event, Lock, Thread, get_ident
 from unittest.mock import Mock, patch
 
 import pytest
 
 from cassandra.cluster import Cluster, ControlConnection
-from cassandra._lifecycle import LifecycleCoordinator, PoolFailureReservation
+from cassandra._lifecycle import (
+    ConnectionFailureReservation, LifecycleCoordinator, PoolFailureReservation)
 from cassandra.connection import (
     Connection, ConnectionException, DefaultEndPoint)
 from cassandra.policies import HostDistance
@@ -964,6 +965,261 @@ def close(self):
             self.is_closed = True
 
 
+def test_duplicate_failed_callback_allows_same_pool_use_response(topology):
+    from cassandra.protocol import ResultMessage, RESULT_KIND_SET_KEYSPACE
+    from cassandra.shard_info import _ShardingInfo
+
+    cluster, _ = topology
+    host = cluster.metadata.get_host_by_host_id(HOST_ID_3)
+    session = new_session_with_pool_state()
+    session.cluster = cluster
+    session.keyspace = None
+    session.is_shutdown = False
+    session.submit = Mock()
+    original = _PublicationConnection(host.endpoint, compression=False)
+    original.features = ProtocolFeatures(shard_id=0)
+    candidate = _PublicationConnection(host.endpoint, compression=False)
+    candidate.features = ProtocolFeatures(shard_id=0)
+    cluster.connection_factory = Mock(side_effect=[original, candidate])
+    cluster.shard_aware_options.disable_shardaware_port = True
+    pool = HostConnection(host, HostDistance.LOCAL, session)
+    session._pools[host] = pool
+    session.keyspace = pool._keyspace = original.keyspace = 'ks'
+    host.sharding_info = _ShardingInfo(0, 2, '', '', 0, None, None)
+    host.signal_connection_failure = Mock(return_value=False)
+    original.in_flight = 2
+    original.is_defunct = True
+    original.last_error = ConnectionException('socket failed', host.endpoint)
+    pool.return_connection(original)
+    assert original.in_flight == 1
+    session.submit.assert_called_once_with(pool._replace, original)
+    pool._replace(original)
+    assert session.submit.call_count == 2
+    session.submit.assert_called_with(pool._open_connection_to_missing_shard, 0)
+
+    io_waiting = Event()
+    reactor_done = Event()
+    response_lock = Lock()
+    responses = []
+    errors = []
+
+    def send(message, request_id, cb, **kwargs):
+        with response_lock:
+            responses.append(cb)
+        io_waiting.set()
+        return 0
+
+    def respond():
+        with response_lock:
+            callbacks = list(responses)
+            responses.clear()
+        for callback in callbacks:
+            callback(ResultMessage(RESULT_KIND_SET_KEYSPACE))
+
+    def run(fn):
+        try:
+            fn()
+        except BaseException as exc:
+            errors.append(exc)
+
+    def process_responses():
+        # The same reactor must finish this redundant failure callback before
+        # delivering the USE response that releases the replacement work lock.
+        pool.return_connection(original)
+        respond()
+        reactor_done.set()
+
+    candidate.send_msg = send
+    worker = Thread(target=run, args=(lambda: pool._open_connection_to_missing_shard(0),))
+    reactor = Thread(target=run, args=(process_responses,))
+    try:
+        worker.start()
+        assert io_waiting.wait(5)
+        reactor.start()
+        assert reactor_done.wait(2), 'duplicate failure blocked same-pool USE response'
+    finally:
+        respond()
+        for thread in (worker, reactor):
+            if thread.ident is not None:
+                thread.join(5)
+        pool.shutdown()
+    assert all(not thread.is_alive() for thread in (worker, reactor))
+    assert not errors
+    assert original.in_flight == candidate.in_flight == 0
+    assert candidate.keyspace == 'ks'
+    host.signal_connection_failure.assert_called_once_with(original.last_error)
+    assert session.submit.call_count == 2
+
+
+@pytest.mark.parametrize('independent_healthy_pool', [False, True])
+def test_concurrent_session_failures_retire_before_health_inspection(
+        topology, independent_healthy_pool):
+    cluster, _ = topology
+    host = cluster.metadata.get_host_by_host_id(HOST_ID_3)
+    sessions, pools, failed, siblings = [], [], [], []
+    for index in range(2 + int(independent_healthy_pool)):
+        session = new_session_with_pool_state()
+        session.cluster = cluster
+        session.keyspace = None
+        session.is_shutdown = False
+        session.shutdown = Mock()
+        session.on_down = Mock()
+        connection = _PublicationConnection(host.endpoint, compression=False)
+        connection.features = ProtocolFeatures(shard_id=0)
+        cluster.connection_factory = Mock(return_value=connection)
+        pool = HostConnection(host, HostDistance.LOCAL, session)
+        session._pools[host] = pool
+        sessions.append(session)
+        pools.append(pool)
+        if index < 2:
+            sibling = _PublicationConnection(host.endpoint, compression=False)
+            sibling.features = ProtocolFeatures(shard_id=1)
+            pool._connections[1] = sibling
+            siblings.append(sibling)
+            connection.is_defunct = True
+            connection.in_flight = 1
+            connection.last_error = ConnectionException('shard failed', host.endpoint)
+            failed.append(connection)
+    cluster.sessions = tuple(sessions)
+    host.signal_connection_failure = Mock(return_value=True)
+    cluster.control_connection.on_down = Mock()
+    snapshots = Barrier(2)
+    for session in sessions[:2]:
+        get_state = session.get_pool_state
+
+        def inspect(get_state=get_state):
+            state = get_state()
+            # Both failures capture the other session's real pool health
+            # before either can finish DOWN admission or its old cleanup.
+            snapshots.wait(timeout=5)
+            return state
+
+        session.get_pool_state = inspect
+
+    errors, scheduled, order = [], [], []
+    cluster.profile_manager.on_down.side_effect = lambda _: order.append('down')
+    cluster._start_reconnector.side_effect = lambda *a, **kw: order.append('recovery')
+
+    def delayed_submit(fn, *args, **kwargs):
+        scheduled.append((fn, args, kwargs))
+        return Future()
+
+    def fail(index):
+        try:
+            pools[index].return_connection(failed[index])
+        except BaseException as exc:
+            errors.append(exc)
+
+    workers = [Thread(target=fail, args=(index,)) for index in range(2)]
+    try:
+        with patch.object(cluster.executor, 'submit', side_effect=delayed_submit):
+            for worker in workers:
+                worker.start()
+            for worker in workers:
+                worker.join(7)
+            assert all(not worker.is_alive() for worker in workers)
+            assert not errors
+            assert all(pool.is_shutdown for pool in pools[:2])
+            assert all(connection.is_closed for connection in failed + siblings)
+            assert all(connection.in_flight == 0 for connection in failed)
+            assert host.is_up is independent_healthy_pool
+            assert len(scheduled) == int(not independent_healthy_pool)
+            # Check actual FIFO delivery and recovery, including a later
+            # notification. A stale second failure must not duplicate DOWN.
+            with cluster._lifecycle.transaction(drain=False):
+                cluster._lifecycle.submit(order.append, 'later')
+            cluster._lifecycle.drain()
+            for fn, args, kwargs in scheduled:
+                fn(*args, **kwargs)
+            assert order == (['later'] if independent_healthy_pool else
+                             ['down', 'recovery', 'later'])
+            assert cluster._start_reconnector.call_count == int(not independent_healthy_pool)
+            if independent_healthy_pool:
+                healthy, _ = pools[2].borrow_connection(timeout=0)
+                assert not healthy.is_closed
+                pools[2].return_connection(healthy)
+                assert not pools[2].is_shutdown
+    finally:
+        snapshots.abort()
+        for worker in workers:
+            if worker.ident is not None:
+                worker.join(5)
+        for pool in pools:
+            pool.shutdown()
+
+
+@pytest.mark.parametrize('supersede', [
+    None, 'pool', 'endpoint', 'host', 'session', 'cluster', 'health',
+])
+def test_retired_failure_admission_does_not_wait_for_pool_work(
+        topology, supersede):
+    cluster, _ = topology
+    host = cluster.metadata.get_host_by_host_id(HOST_ID_3)
+    session = new_session_with_pool_state()
+    session.cluster = cluster
+    session.keyspace = None
+    session.is_shutdown = False
+    session.shutdown = Mock()
+    connection = _PublicationConnection(host.endpoint, compression=False)
+    connection.features = ProtocolFeatures(shard_id=0)
+    cluster.connection_factory = Mock(return_value=connection)
+    pool = HostConnection(host, HostDistance.LOCAL, session)
+    session._pools[host] = pool
+    cluster.sessions = (session,)
+    successor = HostConnection(host, HostDistance.IGNORED, session)
+    reservation = ConnectionFailureReservation(pool, connection)
+    pool.shutdown(_failure=reservation)
+    cluster.on_down_potentially_blocking = Mock(return_value=Future())
+
+    def distance(_host):
+        with session._lock, host.lock:
+            if supersede == 'pool':
+                session._pools[host] = successor
+            elif supersede == 'endpoint':
+                host.endpoint = DefaultEndPoint('127.0.0.2')
+            elif supersede == 'host':
+                host._is_removed = True
+            elif supersede == 'session':
+                session.is_shutdown = True
+            elif supersede == 'cluster':
+                cluster.is_shutdown = True
+            elif supersede == 'health':
+                assert cluster._lifecycle.mark_ready(host)
+        return HostDistance.LOCAL
+
+    cluster.profile_manager.distance.side_effect = distance
+    done = Event()
+    admitted, errors = [], []
+
+    def admit():
+        try:
+            admitted.append(cluster.on_down(host, False, _pool_failure=reservation))
+        except BaseException as exc:
+            errors.append(exc)
+        finally:
+            done.set()
+
+    worker = Thread(target=admit)
+    try:
+        # After retirement the shard map has no authority left to inspect.
+        # Its work lock must not delay admission or hide superseding topology.
+        with pool._lock:
+            worker.start()
+            assert done.wait(2), 'retired failure waited for pool work'
+        worker.join(5)
+        assert not worker.is_alive()
+        assert not errors
+        assert admitted == [supersede is None]
+        assert host.is_up is (supersede is not None)
+        assert cluster.on_down_potentially_blocking.call_count == int(supersede is None)
+        assert not successor.is_shutdown
+    finally:
+        worker.join(5)
+        cluster.is_shutdown = False
+        pool.shutdown()
+        successor.shutdown()
+
+
 @pytest.mark.parametrize('failure', ['close', 'defunct'])
 def test_failed_candidate_cannot_fence_down_before_publication(
         topology, failure):
@@ -1191,12 +1447,9 @@ def publish():
     host.signal_connection_failure.assert_not_called()
 
 
-@pytest.mark.parametrize('phase, supersede', [
-    ('return', None), ('down', None), ('return', 'pool'),
-    ('down', 'pool'), ('down', 'health'),
-])
+@pytest.mark.parametrize('supersede', [None, 'pool'])
 def test_current_failure_contention_allows_other_host_response(
-        topology, phase, supersede):
+        topology, supersede):
     from cassandra.protocol import ResultMessage, RESULT_KIND_SET_KEYSPACE
     from cassandra.shard_info import _ShardingInfo
 
@@ -1229,7 +1482,6 @@ def test_current_failure_contention_allows_other_host_response(
         host.signal_connection_failure = Mock(return_value=True)
 
     io_waiting = Event()
-    at_down = Event()
     contended = Event()
     reactor_done = Event()
     response_lock = Lock()
@@ -1260,15 +1512,6 @@ def admit(host, is_addition, generation):
         return future
 
     cluster.on_down_potentially_blocking = Mock(side_effect=admit)
-    begin_down = cluster._lifecycle.begin_down
-
-    def begin(host, *args, **kwargs):
-        if phase == 'down' and host is hosts[0]:
-            at_down.set()
-            assert io_waiting.wait(5)
-        return begin_down(host, *args, **kwargs)
-
-    cluster._lifecycle.begin_down = begin
     work_lock = pools[0]._lock
 
     class ObservedLock:
@@ -1294,33 +1537,24 @@ def run(fn):
         except BaseException as exc:
             errors.append(exc)
 
-    def open_shard():
-        if phase == 'down':
-            assert at_down.wait(5)
-        pools[0]._open_connection_to_missing_shard(1)
-
     def process_responses():
         # One reactor must finish the other host's failure callback before
         # it can deliver the USE response that releases the pool work lock.
         if supersede == 'pool':
             with session._lock, hosts[0].lock:
                 session._pools[hosts[0]] = successor
-        elif supersede == 'health':
-            assert cluster._lifecycle.mark_ready(hosts[0])
         pools[1].return_connection(connections[1])
         respond()
         reactor_done.set()
 
     failure = Thread(target=run, args=(lambda: pools[0].return_connection(
         connections[0]),))
-    worker = Thread(target=run, args=(open_shard,))
+    worker = Thread(target=run, args=(lambda: pools[0]._open_connection_to_missing_shard(1),))
     reactor = Thread(target=run, args=(process_responses,))
     try:
         worker.start()
-        if phase == 'return':
-            assert io_waiting.wait(5)
-        failure.start()
         assert io_waiting.wait(5)
+        failure.start()
         assert contended.wait(5)
         reactor.start()
         assert reactor_done.wait(2), 'cross-host failure blocked reactor progress'
@@ -1342,8 +1576,7 @@ def process_responses():
     assert hosts[0].is_up is bool(supersede)
     assert not hosts[1].is_up
     hosts[1].signal_connection_failure.assert_called_once()
-    assert hosts[0].signal_connection_failure.call_count == int(
-        phase == 'down' or supersede != 'pool')
+    assert hosts[0].signal_connection_failure.call_count == int(supersede != 'pool')
     if successor is not None:
         assert session._pools[hosts[0]] is successor
         assert not successor.is_shutdown

From c9a3fb287d1e5a28f1eb9dc70c97cc1ee7ab3962 Mon Sep 17 00:00:00 2001
From: Dmitry Kropachev <dmitry.kropachev@gmail.com>
Date: Sun, 25 Oct 2026 04:47:41 -0400
Subject: [PATCH 6/7] pool: claim socket failures before waiting for pool work

A missing-shard worker can hold pool work while waiting for USE. If
the first failed callback waits before marking the socket handled,
a redundant reactor callback also waits and cannot deliver the response.

Serialize the existing duplicate predicate and handled marker under
the connection lock before waiting. Retain reservation checks before
policy, DOWN and replacement effects, and release the connection lock
before those checks.

Cover both an already-handled callback and one whose first failure
is already waiting, preserving accounting and one policy call.
---
 cassandra/pool.py                         | 23 +++---
 tests/unit/test_host_connection_pool.py   | 51 ++++--------
 tests/unit/test_lifecycle_coordination.py | 96 +++++++++++++++++------
 3 files changed, 99 insertions(+), 71 deletions(-)

diff --git a/cassandra/pool.py b/cassandra/pool.py
index 187c55e033..63a4827fe1 100644
--- a/cassandra/pool.py
+++ b/cassandra/pool.py
@@ -655,22 +655,19 @@ def return_connection(self, connection, stream_was_orphaned=False,
                 self._stream_available_condition.notify()
 
         if connection.is_defunct or connection.is_closed:
-            if connection.signaled_error and not replace_only and \
-                    (not self.shutdown_on_error or
-                     (isinstance(connection.last_error, ConnectionSendError) and
-                      not heartbeat_failed)):
-                return
+            # Claim the transport before pool work: a redundant reactor
+            # callback must return while that work waits for its response.
+            with connection.lock:
+                if connection.signaled_error and not replace_only and \
+                        (not self.shutdown_on_error or
+                         (isinstance(connection.last_error, ConnectionSendError) and
+                          not heartbeat_failed)):
+                    return
+                report_failure = not replace_only and not connection.signaled_error
+                connection.signaled_error = True
             reservation = _lifecycle.ConnectionFailureReservation(self, connection)
             with reservation.locked():
                 current = reservation.owns_locked()
-                if current:
-                    if connection.signaled_error and not replace_only and \
-                            (not self.shutdown_on_error or
-                             (isinstance(connection.last_error, ConnectionSendError) and
-                              not heartbeat_failed)):
-                        return
-                    report_failure = not replace_only and not connection.signaled_error
-                    connection.signaled_error = True
             if not current:
                 with self._lock:
                     self._trash.discard(connection)
diff --git a/tests/unit/test_host_connection_pool.py b/tests/unit/test_host_connection_pool.py
index 62e32028a0..34bd57737b 100644
--- a/tests/unit/test_host_connection_pool.py
+++ b/tests/unit/test_host_connection_pool.py
@@ -12,7 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 from concurrent.futures import Future, ThreadPoolExecutor
-from contextlib import contextmanager
 import logging
 import uuid
 from cassandra.protocol_features import ProtocolFeatures
@@ -20,7 +19,7 @@
 from cassandra.shard_info import _ShardingInfo
 
 import unittest
-from threading import Barrier, Thread, Event, Lock, RLock, get_ident
+from threading import Barrier, Thread, Event, Lock, RLock
 from unittest.mock import Mock, NonCallableMagicMock, MagicMock, patch
 
 from cassandra.cluster import Cluster, Session, ShardAwareOptions
@@ -140,7 +139,7 @@ def replace():
         try:
             winner = (HostConnection(host, HostDistance.LOCAL, session)
                       if change == 'pool' else None)
-            with session._lock, host.lock, pool._lock:
+            with original.lock, session._lock, host.lock, pool._lock:
                 if change == 'socket':
                     pool._connections[0] = replacement
                     pool._trash.add(original)
@@ -292,58 +291,42 @@ def test_racing_first_callbacks_preserve_convicting_owner(socket_pool):
     host, session, pool, original, _ = socket_pool
     original.in_flight = 2
     original.is_defunct = True
-    initial_callbacks = Barrier(2)
-    policy_entered, release_policy, second_done = Event(), Event(), Event()
+    start_callbacks = Barrier(3)
+    policy_entered, release_policy, duplicate_done = Event(), Event(), Event()
     errors = []
-    locked = ConnectionFailureReservation.locked
-
-    @contextmanager
-    def enter(reservation, *args, **kwargs):
-        if not getattr(reservation, '_test_entered', False):
-            reservation._test_entered = True
-            # Both callbacks have passed the unsignaled fast check before
-            # the first callback publishes its conviction-in-progress.
-            initial_callbacks.wait(timeout=5)
-            if get_ident() == second.ident:
-                assert policy_entered.wait(5)
-        with locked(reservation, *args, **kwargs):
-            yield
 
     def convict(error):
         policy_entered.set()
         assert release_policy.wait(5)
         return True
 
-    def fail(done=None):
+    def fail():
         try:
+            start_callbacks.wait(timeout=5)
             pool.return_connection(original)
         except BaseException as exc:
             errors.append(exc)
         finally:
-            if done is not None:
-                done.set()
+            duplicate_done.set()
 
     host.signal_connection_failure.side_effect = convict
-    first = Thread(target=fail)
-    second = Thread(target=fail, args=(second_done,))
+    workers = (Thread(target=fail), Thread(target=fail))
     try:
-        with patch.object(ConnectionFailureReservation, 'locked', enter):
-            first.start()
-            second.start()
-            assert second_done.wait(2)
-            assert pool._connections[0] is original
-            release_policy.set()
-            for worker in (first, second):
-                worker.join(5)
+        for worker in workers:
+            worker.start()
+        start_callbacks.wait(timeout=5)
+        assert policy_entered.wait(5)
+        assert duplicate_done.wait(2), 'duplicate callback waited for conviction'
+        assert pool._connections[0] is original
     finally:
         release_policy.set()
-        initial_callbacks.abort()
-        for worker in (first, second):
+        start_callbacks.abort()
+        for worker in workers:
             if worker.ident is not None:
                 worker.join(5)
         pool.shutdown()
     assert not errors
-    assert all(not worker.is_alive() for worker in (first, second))
+    assert all(not worker.is_alive() for worker in workers)
     assert original.in_flight == 0
     host.signal_connection_failure.assert_called_once()
     session.cluster.on_down.assert_called_once()
diff --git a/tests/unit/test_lifecycle_coordination.py b/tests/unit/test_lifecycle_coordination.py
index 447d218b7b..2497b27582 100644
--- a/tests/unit/test_lifecycle_coordination.py
+++ b/tests/unit/test_lifecycle_coordination.py
@@ -965,7 +965,10 @@ def close(self):
             self.is_closed = True
 
 
-def test_duplicate_failed_callback_allows_same_pool_use_response(topology):
+@pytest.mark.parametrize('first_failure_waiting', [False, True],
+                         ids=['already_claimed', 'first_waiting'])
+def test_duplicate_failed_callback_allows_same_pool_use_response(
+        topology, first_failure_waiting):
     from cassandra.protocol import ResultMessage, RESULT_KIND_SET_KEYSPACE
     from cassandra.shard_info import _ShardingInfo
 
@@ -977,7 +980,7 @@ def test_duplicate_failed_callback_allows_same_pool_use_response(topology):
     session.is_shutdown = False
     session.submit = Mock()
     original = _PublicationConnection(host.endpoint, compression=False)
-    original.features = ProtocolFeatures(shard_id=0)
+    original.features = ProtocolFeatures(shard_id=1 if first_failure_waiting else 0)
     candidate = _PublicationConnection(host.endpoint, compression=False)
     candidate.features = ProtocolFeatures(shard_id=0)
     cluster.connection_factory = Mock(side_effect=[original, candidate])
@@ -988,22 +991,48 @@ def test_duplicate_failed_callback_allows_same_pool_use_response(topology):
     host.sharding_info = _ShardingInfo(0, 2, '', '', 0, None, None)
     host.signal_connection_failure = Mock(return_value=False)
     original.in_flight = 2
-    original.is_defunct = True
-    original.last_error = ConnectionException('socket failed', host.endpoint)
-    pool.return_connection(original)
-    assert original.in_flight == 1
-    session.submit.assert_called_once_with(pool._replace, original)
-    pool._replace(original)
-    assert session.submit.call_count == 2
-    session.submit.assert_called_with(pool._open_connection_to_missing_shard, 0)
+    original.last_error = ConnectionResetError('socket reset')
+    if not first_failure_waiting:
+        original.is_defunct = True
+        pool.return_connection(original)
+        assert original.in_flight == 1
+        session.submit.assert_called_once_with(pool._replace, original)
+        pool._replace(original)
+        assert session.submit.call_count == 2
+        session.submit.assert_called_with(pool._open_connection_to_missing_shard, 0)
 
     io_waiting = Event()
+    first_blocking = Event()
     reactor_done = Event()
     response_lock = Lock()
     responses = []
     errors = []
 
+    if first_failure_waiting:
+        work_lock = pool._lock
+
+        class WorkLock:
+            def acquire(self, blocking=True):
+                # A nonblocking ownership probe is not the common checkpoint:
+                # the first callback must actually wait behind pending USE.
+                if blocking and get_ident() == first.ident:
+                    first_blocking.set()
+                return work_lock.acquire(blocking)
+
+            def release(self):
+                work_lock.release()
+
+            def __enter__(self):
+                self.acquire()
+                return self
+
+            def __exit__(self, *args):
+                self.release()
+
+        pool._lock = WorkLock()
+
     def send(message, request_id, cb, **kwargs):
+        assert message.query.startswith('USE ')
         with response_lock:
             responses.append(cb)
         io_waiting.set()
@@ -1031,24 +1060,43 @@ def process_responses():
 
     candidate.send_msg = send
     worker = Thread(target=run, args=(lambda: pool._open_connection_to_missing_shard(0),))
+    first = Thread(target=run, args=(lambda: pool.return_connection(original),))
     reactor = Thread(target=run, args=(process_responses,))
+    threads = (worker, first, reactor) if first_failure_waiting else (worker, reactor)
     try:
-        worker.start()
-        assert io_waiting.wait(5)
-        reactor.start()
-        assert reactor_done.wait(2), 'duplicate failure blocked same-pool USE response'
+        try:
+            worker.start()
+            assert io_waiting.wait(5)
+            if first_failure_waiting:
+                original.is_defunct = True
+                first.start()
+                assert first_blocking.wait(5), 'first failure did not wait for USE'
+            reactor.start()
+            progressed = reactor_done.wait(2)
+        finally:
+            respond()
+            for thread in threads:
+                if thread.ident is not None:
+                    thread.join(5)
+        assert all(not thread.is_alive() for thread in threads)
+        assert not errors
+        assert original.in_flight == candidate.in_flight == 0
+        assert candidate.keyspace == 'ks'
+        host.signal_connection_failure.assert_called_once_with(original.last_error)
+        assert host.is_up
+        assert not pool.is_shutdown
+        if first_failure_waiting:
+            session.submit.assert_called_once_with(pool._replace, original)
+            assert session._pools[host] is pool
+            borrowed, _ = pool.borrow_connection(timeout=0)
+            assert borrowed is candidate
+            pool.return_connection(borrowed)
+            assert candidate.in_flight == 0
+        else:
+            assert session.submit.call_count == 2
+        assert progressed, 'duplicate failure blocked same-pool USE response'
     finally:
-        respond()
-        for thread in (worker, reactor):
-            if thread.ident is not None:
-                thread.join(5)
         pool.shutdown()
-    assert all(not thread.is_alive() for thread in (worker, reactor))
-    assert not errors
-    assert original.in_flight == candidate.in_flight == 0
-    assert candidate.keyspace == 'ks'
-    host.signal_connection_failure.assert_called_once_with(original.last_error)
-    assert session.submit.call_count == 2
 
 
 @pytest.mark.parametrize('independent_healthy_pool', [False, True])

From 41b16041c62f70ad0be37d290e5cb874ab58958b Mon Sep 17 00:00:00 2001
From: Dmitry Kropachev <dmitry.kropachev@gmail.com>
Date: Sun, 25 Oct 2026 05:29:19 -0400
Subject: [PATCH 7/7] tests: settle late USE callbacks during teardown

A worker may register its USE callback after the response queue has
been drained during failed setup. The bounded join then leaves it
waiting under the pool work lock, and shutdown can hang.

Publish cleanup state under the response queue lock and immediately
settle callbacks registered afterward. Keep callback execution outside
the lock and preserve the existing controlled response ordering.

Delayed-send controls now fail the original assertion and exit cleanly.
---
 tests/unit/test_lifecycle_coordination.py | 26 +++++++++++++++++------
 1 file changed, 20 insertions(+), 6 deletions(-)

diff --git a/tests/unit/test_lifecycle_coordination.py b/tests/unit/test_lifecycle_coordination.py
index 2497b27582..fe1927f408 100644
--- a/tests/unit/test_lifecycle_coordination.py
+++ b/tests/unit/test_lifecycle_coordination.py
@@ -1006,6 +1006,7 @@ def test_duplicate_failed_callback_allows_same_pool_use_response(
     reactor_done = Event()
     response_lock = Lock()
     responses = []
+    cleanup_started = False
     errors = []
 
     if first_failure_waiting:
@@ -1034,12 +1035,18 @@ def __exit__(self, *args):
     def send(message, request_id, cb, **kwargs):
         assert message.query.startswith('USE ')
         with response_lock:
-            responses.append(cb)
+            respond_now = cleanup_started
+            if not respond_now:
+                responses.append(cb)
+        if respond_now:
+            cb(ResultMessage(RESULT_KIND_SET_KEYSPACE))
         io_waiting.set()
         return 0
 
-    def respond():
+    def respond(cleanup=False):
+        nonlocal cleanup_started
         with response_lock:
+            cleanup_started |= cleanup
             callbacks = list(responses)
             responses.clear()
         for callback in callbacks:
@@ -1074,7 +1081,7 @@ def process_responses():
             reactor.start()
             progressed = reactor_done.wait(2)
         finally:
-            respond()
+            respond(cleanup=True)
             for thread in threads:
                 if thread.ident is not None:
                     thread.join(5)
@@ -1534,20 +1541,27 @@ def test_current_failure_contention_allows_other_host_response(
     reactor_done = Event()
     response_lock = Lock()
     responses = []
+    cleanup_started = False
     delivered = []
     delivery_futures = []
     errors = []
 
     def send(message, request_id, cb, **kwargs):
         with response_lock:
-            responses.append(cb)
+            respond_now = cleanup_started
+            if not respond_now:
+                responses.append(cb)
+        if respond_now:
+            cb(ResultMessage(RESULT_KIND_SET_KEYSPACE))
         io_waiting.set()
         return 0
 
     candidate.send_msg = send
 
-    def respond():
+    def respond(cleanup=False):
+        nonlocal cleanup_started
         with response_lock:
+            cleanup_started |= cleanup
             callbacks = list(responses)
             responses.clear()
         for callback in callbacks:
@@ -1609,7 +1623,7 @@ def process_responses():
     finally:
         # Release controlled I/O even for the unfixed production negative
         # control, so assertion failures cannot leave blocked driver threads.
-        respond()
+        respond(cleanup=True)
         for thread in (worker, failure, reactor):
             if thread.ident is not None:
                 thread.join(5)