Queued DOWN handling can reconnect a host after removal

[dkropachev]

Problem

Cluster.on_down() submits the blocking part of DOWN processing to the executor:

python-driver/cassandra/cluster.py

Lines 2013 to 2052 in f0004a8

	@run_in_executor
	def on_down_potentially_blocking(self, host, is_host_addition):
	self.profile_manager.on_down(host)
	self.control_connection.on_down(host)
	for session in tuple(self.sessions):
	session.on_down(host)
	for listener in self.listeners:
	listener.on_down(host)
	self._start_reconnector(host, is_host_addition)
	def on_down(self, host, is_host_addition, expect_host_to_be_down=False):
	"""
	Intended for internal use only.
	"""
	if self.is_shutdown or self.allow_control_connection_query_fallback == ControlConnectionQueryFallback.SkipPoolCreation:
	return
	with host.lock:
	was_up = host.is_up
	# ignore down signals if we have open pools to the host
	# this is to avoid closing pools when a control connection host became isolated
	if self._discount_down_events and self.profile_manager.distance(host) != HostDistance.IGNORED:
	connected = False
	for session in tuple(self.sessions):
	pool_states = session.get_pool_state()
	pool_state = pool_states.get(host)
	if pool_state:
	connected |= pool_state['open_count'] > 0
	if connected:
	return
	host.set_down()
	if (not was_up and not expect_host_to_be_down) or host.is_currently_reconnecting():
	return
	log.warning("Host %s has been marked down", host)
	self.on_down_potentially_blocking(host, is_host_addition)

The host can be removed after that work is submitted but before it runs. The removal lifecycle cancels the reconnection handler currently installed on the host:

python-driver/cassandra/cluster.py

Lines 2124 to 2139 in f0004a8

	def on_remove(self, host):
	if self.is_shutdown:
	return
	log.debug("[cluster] Removing host %s", host)
	host.set_down()
	self.profile_manager.on_remove(host)
	for session in tuple(self.sessions):
	session.on_remove(host)
	for listener in self.listeners:
	listener.on_remove(host)
	self.control_connection.on_remove(host)
	reconnection_handler = host.get_and_set_reconnection_handler(None)
	if reconnection_handler:
	reconnection_handler.cancel()

If the queued DOWN task has not yet created its handler, there is nothing for removal to cancel. When the task eventually runs, it unconditionally sends DOWN notifications and starts a new reconnector:

python-driver/cassandra/cluster.py

Lines 2013 to 2023 in f0004a8

	@run_in_executor
	def on_down_potentially_blocking(self, host, is_host_addition):
	self.profile_manager.on_down(host)
	self.control_connection.on_down(host)
	for session in tuple(self.sessions):
	session.on_down(host)
	for listener in self.listeners:
	listener.on_down(host)
	self._start_reconnector(host, is_host_addition)

_start_reconnector() does not verify that the Host is still registered before installing and starting the handler:

python-driver/cassandra/cluster.py

Lines 1989 to 2011 in f0004a8

	def _start_reconnector(self, host, is_host_addition):
	if self.profile_manager.distance(host) == HostDistance.IGNORED:
	return
	schedule = self.reconnection_policy.new_schedule()
	# in order to not hold references to this Cluster open and prevent
	# proper shutdown when the program ends, we'll just make a closure
	# of the current Cluster attributes to create new Connections with
	conn_factory = self._make_connection_factory(host)
	reconnector = _HostReconnectionHandler(
	host, conn_factory, is_host_addition, self.on_add, self.on_up,
	self.scheduler, schedule, host.get_and_set_reconnection_handler,
	new_handler=None)
	old_reconnector = host.get_and_set_reconnection_handler(reconnector)
	if old_reconnector:
	log.debug("Old host reconnector found for %s, cancelling", host)
	old_reconnector.cancel()
	log.debug("Starting reconnector for host %s", host)
	reconnector.start()

A removed host can therefore acquire a reconnector after its removal lifecycle has completed. A successful attempt can subsequently run the UP lifecycle and recreate session state for a host that is no longer present in cluster metadata.

Reproduction

The race can be reproduced deterministically by controlling executor execution:

1. Register and mark a host up.
2. Call Cluster.on_down() while holding its executor task pending.
3. Remove the host before allowing the DOWN task to run.
4. Run the queued task.
5. Observe that it starts a reconnection handler for the removed host.

Expected behavior

Once a host has completed removal, previously queued DOWN work must not:

• notify policies, sessions, or listeners about another transition;
• install a new reconnection handler;
• run the host UP lifecycle after a successful connection.

Host registration checks must use the identity of the exact Host object rather than assuming its current host_id is already the metadata key, because topology refresh can temporarily reindex the same object.

Acceptance criteria

• A queued DOWN task does not start a reconnector after its host is removed.
• _start_reconnector() cannot install a handler after concurrent host removal completes.
• No policy, session, control-connection, or listener DOWN notification is emitted by stale work after removal.
• A host that is merely being reindexed, rather than removed, is not mistaken for an unregistered host.
• Unit tests reproduce the relevant interleavings without timing-dependent sleeps.

Related

• Route topology-discovered host removals through the cluster lifecycle #1007 tracks topology reconciliation bypassing the cluster removal lifecycle.
• Epic: Host identity, topology immutability, and pool lifecycle refactor #921 tracks the broader host-identity and topology-lifecycle refactor.
• Add topology-generation fencing for pools and connections #925 tracks generation fencing for stale asynchronous topology work.