host: make host_id the canonical identity

CHANGELOG.rst

@@ -31,6 +31,11 @@ Features
 
 Others
 ------
+* ``Host.host_id`` is now the immutable identity of a node. Constructing a
+  ``Host`` requires a non-nil ``uuid.UUID``; equality, hashing, and ordering use
+  only that ID, and comparing a ``Host`` with an address no longer reports them
+  as equal. Because hashing is stable, existing Host-keyed session pool
+  lookups and removals remain valid across endpoint changes (issue #867).
 * ``DCAwareRoundRobinPolicy.local_dc`` is now read-only. It is set by the constructor,
   and filled in by the policy itself when the constructor was given none, from the first
   host to come up. Assigning it afterwards was indistinguishable from that inference,

cassandra/metrics.py

@@ -441,7 +441,9 @@ def __init__(self, cluster_proxy):
             Stat('known_hosts',
                 lambda: len(cluster_proxy.metadata.all_hosts())),
             Stat('connected_to',
-                lambda: len(set(chain.from_iterable(list(s._pools.keys()) for s in cluster_proxy.sessions)))),
+                lambda: len(set(chain.from_iterable(
+                    (pool.host for pool in list(s._pools.values()))
+                    for s in cluster_proxy.sessions)))),
             Stat('open_connections',
                 lambda: sum(sum(p.open_count for p in list(s._pools.values())) for s in cluster_proxy.sessions)))
 

cassandra/pool.py

@@ -21,6 +21,7 @@
 import time
 import random
 import copy
+import uuid
 from threading import Lock, RLock, Condition
 import weakref
 try:
@@ -131,11 +132,6 @@ class Host(object):
     release_version as queried from the control connection system tables
     """
 
-    host_id = None
-    """
-    The unique identifier of the cassandra node
-    """
-
     dse_version = None
     """
     dse_version as queried from the control connection system tables. Only populated when connecting to
@@ -157,6 +153,7 @@ class Host(object):
     Not queried if :attr:`~.Cluster.token_metadata_enabled` is ``False``.
     """
 
+    _host_id = None
     _datacenter = None
     _rack = None
     _reconnection_handler = None
@@ -172,11 +169,15 @@ def __init__(self, endpoint, conviction_policy_factory, datacenter=None, rack=No
         if conviction_policy_factory is None:
             raise ValueError("conviction_policy_factory may not be None")
 
+        if not isinstance(host_id, uuid.UUID):
+            raise TypeError("host_id must be a uuid.UUID")
+        if host_id.int == 0:
+            raise ValueError("host_id may not be the nil UUID")
+
         self.endpoint = endpoint if isinstance(endpoint, EndPoint) else DefaultEndPoint(endpoint)
+        self._host_id = host_id
+        self._is_removed = False
         self.conviction_policy = conviction_policy_factory(self)
-        if not host_id:
-            raise ValueError("host_id may not be None")
-        self.host_id = host_id
         self.set_location_info(datacenter, rack)
         self.lock = RLock()
 
@@ -188,6 +189,13 @@ def address(self):
         # backward compatibility
         return self.endpoint.address
 
+    @property
+    def host_id(self):
+        """
+        The immutable unique identifier of the Cassandra node.
+        """
+        return self._host_id
+
     @property
     def datacenter(self):
         """ The datacenter the node is in.  """
@@ -232,23 +240,25 @@ def get_and_set_reconnection_handler(self, new_handler):
             self._reconnection_handler = new_handler
             return old
 
+    def _clear_reconnection_handler(self, handler):
+        with self.lock:
+            if self._reconnection_handler is not handler:
+                return False
+            self._reconnection_handler = None
+            return not self._is_removed
+
     def __eq__(self, other):
-        if isinstance(other, Host):
-            return self.endpoint == other.endpoint
-        else:  # TODO Backward compatibility, remove next major
-            return self.endpoint.address == other
+        if not isinstance(other, Host):
+            return NotImplemented
+        return self.host_id == other.host_id
 
     def __hash__(self):
-        return hash(self.endpoint)
+        return hash(self.host_id)
 
     def __lt__(self, other):
-        self_is_unix = isinstance(self.endpoint, UnixSocketEndPoint)
-        other_is_unix = isinstance(other.endpoint, UnixSocketEndPoint)
-        if self_is_unix != other_is_unix:
-            # Endpoint comparators assume same-kind operands, so partition
-            # Unix and network Hosts before delegating their ordering.
-            return self_is_unix
-        return self.endpoint < other.endpoint
+        if not isinstance(other, Host):
+            return NotImplemented
+        return self.host_id < other.host_id
 
     def __str__(self):
         return str(self.endpoint)
@@ -265,6 +275,7 @@ class _ReconnectionHandler(object):
     """
 
     _cancelled = False
+    _clear_handler_before_reconnection = False
 
     def __init__(self, scheduler, schedule, callback, *callback_args, **callback_kwargs):
         self.scheduler = scheduler
@@ -305,15 +316,23 @@ def run(self):
                     self.scheduler.schedule(next_delay, self.run)
         else:
             if not self._cancelled:
+                if (self._clear_handler_before_reconnection and
+                        not self._release_reconnection_handler()):
+                    return
                 self.on_reconnection(conn)
-                self.callback(*(self.callback_args), **(self.callback_kwargs))
+                if not self._clear_handler_before_reconnection:
+                    self._release_reconnection_handler()
         finally:
             if conn:
                 conn.close()
 
     def cancel(self):
         self._cancelled = True
 
+    def _release_reconnection_handler(self):
+        self.callback(*(self.callback_args), **(self.callback_kwargs))
+        return True
+
     def try_reconnect(self):
         """
         Subclasses must implement this method.  It should attempt to
@@ -349,6 +368,12 @@ def on_exception(self, exc, next_delay):
 
 class _HostReconnectionHandler(_ReconnectionHandler):
 
+    # Host reconnection callbacks can synchronously start another reconnector
+    # when rebuilding pools fails. Clear this handler first so that failure is
+    # not suppressed as already reconnecting and post-callback cleanup cannot
+    # clear the successor.
+    _clear_handler_before_reconnection = True
+
     def __init__(self, host, connection_factory, is_host_addition, on_add, on_up, *args, **kwargs):
         _ReconnectionHandler.__init__(self, *args, **kwargs)
         self.is_host_addition = is_host_addition
@@ -360,6 +385,9 @@ def __init__(self, host, connection_factory, is_host_addition, on_add, on_up, *a
     def try_reconnect(self):
         return self.connection_factory()
 
+    def _release_reconnection_handler(self):
+        return self.host._clear_reconnection_handler(self)
+
     def on_reconnection(self, connection):
         log.info("Successful reconnection to %s, marking node up if it isn't already", self.host)
         if self.is_host_addition:

tests/integration/simulacron/test_cluster.py

@@ -89,12 +89,23 @@ def test_duplicate(self):
         with MockLoggingHandler().set_module_name(cassandra.cluster.__name__) as mock_handler:
             address_column = "rpc_address"
             rows = [
-                {"peer": "127.0.0.1", "data_center": "dc", "host_id": "dontcare1", "rack": "rack1",
-                "release_version": "3.11.4", address_column: "127.0.0.1", "schema_version": "dontcare", "tokens": "1"},
-                {"peer": "127.0.0.2", "data_center": "dc", "host_id": "dontcare2", "rack": "rack1",
-                "release_version": "3.11.4", address_column: "127.0.0.2", "schema_version": "dontcare", "tokens": "2"},
+                {"peer": "127.0.0.1", "data_center": "dc", "host_id": "00000000-0000-0000-0000-000000000001", "rack": "rack1",
+                "release_version": "3.11.4", address_column: "127.0.0.1", "schema_version": "00000000-0000-0000-0000-000000000011", "tokens": ["1"]},
+                {"peer": "127.0.0.2", "data_center": "dc", "host_id": "00000000-0000-0000-0000-000000000002", "rack": "rack1",
+                "release_version": "3.11.4", address_column: "127.0.0.2", "schema_version": "00000000-0000-0000-0000-000000000012", "tokens": ["2"]},
             ]
-            prime_query(ControlConnection._SELECT_PEERS, rows=rows)
+            prime_query(
+                ControlConnection._SELECT_PEERS, rows=rows,
+                column_types={
+                    "peer": "inet",
+                    "data_center": "varchar",
+                    "host_id": "uuid",
+                    "rack": "varchar",
+                    "release_version": "varchar",
+                    address_column: "inet",
+                    "schema_version": "uuid",
+                    "tokens": "set<varchar>",
+                })
 
             cluster = Cluster(protocol_version=PROTOCOL_VERSION, compression=False)
             session = cluster.connect(wait_for_all_pools=True)

tests/integration/simulacron/test_connection.py

@@ -224,7 +224,7 @@ class PatchedRoundRobinPolicy(RoundRobinPolicy):
             # Send always to same host
             def make_query_plan(self, working_keyspace=None, query=None):
                 if query and query.query_string == query_to_prime:
-                    return filter(lambda h: h == query_host, self._live_hosts)
+                    return filter(lambda h: h.address == query_host, self._live_hosts)
                 else:
                     return super(PatchedRoundRobinPolicy, self).make_query_plan()
 

tests/integration/standard/test_shard_aware.py

@@ -152,7 +152,8 @@ def _assert_blocked_node_disconnected(self, node_ip_address, node_port):
             assert active_control_connection.is_closed or active_control_connection.is_defunct
 
         pools = getattr(self.session, '_pools', None) or {}
-        for host, pool in pools.items():
+        for pool in pools.values():
+            host = pool.host
             if host.endpoint.address != node_ip_address or host.endpoint.port != node_port:
                 continue
 

tests/integration/standard/test_tablets_routing_v2.py

@@ -276,7 +276,8 @@ def _any_connection(self):
 
     @staticmethod
     def _all_shard_connections(session):
-        for host, pool in session._pools.items():
+        for pool in session._pools.values():
+            host = pool.host
             for shard, conn in pool._connections.items():
                 yield host, shard, conn
 
@@ -287,9 +288,9 @@ def _wait_for_shard_connections(session, timeout=15):
         deadline = time.time() + timeout
         while time.time() < deadline:
             if all(
-                len(pool._connections) >= (min(host.sharding_info.shards_count, 2)
-                                           if host.sharding_info else 1)
-                for host, pool in session._pools.items()
+                len(pool._connections) >= (min(pool.host.sharding_info.shards_count, 2)
+                                           if pool.host.sharding_info else 1)
+                for pool in session._pools.values()
             ):
                 return
             time.sleep(0.05)
@@ -329,7 +330,8 @@ def _find_replica_wrong_shard(session, tablet):
         (host, owner_shard, wrong_shard, conn) or None if no host has >=2 shards.
         """
         replica_shard = {host_id: shard for host_id, shard in tablet.replicas}
-        for host, pool in session._pools.items():
+        for pool in session._pools.values():
+            host = pool.host
             owner = replica_shard.get(host.host_id)
             if owner is None:
                 continue

tests/unit/advanced/test_policies.py

@@ -79,7 +79,7 @@ def test_target_host_down(self):
         policy = DSELoadBalancingPolicy(RoundRobinPolicy())
         policy.populate(Mock(metadata=ClusterMetaMock({'127.0.0.1': target_host})), hosts)
         query_plan = list(policy.make_query_plan(None, Mock(target_host='127.0.0.1')))
-        assert sorted(query_plan) == hosts
+        assert sorted(query_plan) == sorted(hosts)
 
         target_host.is_up = False
         policy.on_down(target_host)
@@ -96,5 +96,5 @@ def test_target_host_nominal(self):
         policy.populate(Mock(metadata=ClusterMetaMock({'127.0.0.1': target_host})), hosts)
         for _ in range(10):
             query_plan = list(policy.make_query_plan(None, Mock(target_host='127.0.0.1')))
-            assert sorted(query_plan) == hosts
+            assert sorted(query_plan) == sorted(hosts)
             assert query_plan[0] == target_host