test: fix TcpProxy close/race in test_client_routes.py NLB test helper

TcpProxy.stop()/drop_connections() used to close a connection's sockets
directly, from the manager thread, while that connection's own forwarder
thread could still be blocked in select()/recv() on those exact file
descriptors -- a classic close-under-concurrent-user race. Since fds are
process-global, closing them could let the OS silently recycle the fd
number into a brand new connection before the stale forwarder thread's
blocked call unwound, causing it to read/write/close a socket that no
longer belonged to it (observed here as unhandled
"ValueError: file descriptor cannot be a negative integer (-1)" crashes
in _forward_loop once a socket was closed out from under it). stop()
also only ever joined the accept-loop thread, never the per-connection
forwarder threads it had just closed sockets out from under.

Fix, scoped entirely to this test helper (no driver code touched):
- TcpProxy._connections now maps (client_sock, target_sock) -> the
  forwarder thread serving that pair.
- stop()/drop_connections() now shut down (SHUT_RDWR) both sockets --
  safe to do concurrently with a blocked select()/recv(), unlike
  close() -- and then join() every forwarder thread before returning.
  Only the forwarder thread itself ever closes its own sockets now,
  and only after it has fully stopped using them.
- _handle_new_connection starts the forwarder thread before publishing
  it into _connections, so a concurrent stop()/drop_connections() can
  never observe (and try to join) a thread that hasn't started yet.
- NLBEmulator.add_node()/remove_node() now serialize against each
  other via an RLock, so a new proxy/connection can never be created
  while another thread's remove_node() is still tearing one down.

Fixes #948.

Validation:
- Standalone stress harness (no CCM needed) driving concurrent
  clients through TcpProxy while repeatedly calling
  drop_connections()/stop()+restart from another thread: pre-fix,
  40 iterations produced 162 unhandled ValueError crashes; post-fix,
  40 iterations (same parameters) and a follow-up 150-iteration run
  produced zero corruptions/exceptions/leaked threads.
- Full end-to-end runs of
  TestFullNodeReplacementThroughNlb::test_should_survive_full_node_replacement_through_nlb
  against a real CCM cluster, both before and after the fix, to
  check for behavioral regressions and reproduce the reported
  flakiness.

Author: mykaul

tests/integration/standard/test_client_routes.py

@@ -73,7 +73,9 @@ def __init__(self, listen_host, listen_port, target_host, target_port):
         self._running = False
         self._thread = None
         self._lock = threading.Lock()
-        self._connections = set()
+        # Maps (client_sock, target_sock) -> the forwarder thread serving
+        # that pair, so stop()/drop_connections() can join it (see below).
+        self._connections = {}
         self.total_connections = 0
 
     def start(self):
@@ -98,10 +100,7 @@ def stop(self):
                 self._server_sock.close()
             except Exception:
                 pass
-        with self._lock:
-            for csock, tsock in list(self._connections):
-                self._close_pair(csock, tsock)
-            self._connections.clear()
+        self._shutdown_and_join_connections()
         if self._thread:
             self._thread.join(timeout=5)
         log.info("TcpProxy stopped %s:%d", self.listen_host, self.listen_port)
@@ -120,12 +119,43 @@ def retarget(self, new_host, new_port):
 
     def drop_connections(self):
         """Forcibly close all active connections."""
-        with self._lock:
-            for csock, tsock in list(self._connections):
-                self._close_pair(csock, tsock)
-            self._connections.clear()
+        self._shutdown_and_join_connections()
         log.info("TcpProxy %s:%d dropped all connections", self.listen_host, self.listen_port)
 
+    def _shutdown_and_join_connections(self):
+        """
+        Interrupt every active connection's forwarder thread and wait for
+        it to exit before returning.
+
+        This must NOT call socket.close() itself: closing a file
+        descriptor that another thread is (or might still be) blocked on
+        in select()/recv() is a classic close-under-concurrent-user race
+        -- the fd number can be silently reassigned by the OS to a brand
+        new connection before the stale forwarder thread's blocked call
+        unwinds, causing it to read/write/close a completely unrelated
+        socket. socket.shutdown(SHUT_RDWR) is safe to call concurrently
+        with a blocking select()/recv() on the same socket (it makes the
+        blocked call return promptly, without invalidating the fd), so we
+        use that to signal the thread and then join() it -- only the
+        forwarder thread itself (in _forward_loop's `finally`) ever calls
+        close() on its own sockets, and only after it has fully stopped
+        using them.
+        """
+        with self._lock:
+            connections = list(self._connections.items())
+        for (csock, tsock), _thread in connections:
+            self._shutdown_pair(csock, tsock)
+        for (csock, tsock), thread in connections:
+            thread.join(timeout=5)
+            if thread.is_alive():
+                log.warning(
+                    "TcpProxy %s:%d: forwarder thread %s did not exit "
+                    "within timeout; leaked fds are possible",
+                    self.listen_host, self.listen_port, thread.name)
+        with self._lock:
+            for key, _thread in connections:
+                self._connections.pop(key, None)
+
     def _run(self):
         while self._running:
             try:
@@ -153,14 +183,17 @@ def _handle_new_connection(self, client_sock, target_host=None, target_port=None
             client_sock.close()
             return
 
-        with self._lock:
-            self._connections.add((client_sock, target_sock))
-            self.total_connections += 1
-
         t = threading.Thread(target=self._forward_loop,
                              args=(client_sock, target_sock),
                              daemon=True)
+        # Start the thread *before* publishing it to self._connections:
+        # otherwise stop()/drop_connections() could snapshot this entry
+        # and call .join() on a thread that hasn't been started yet,
+        # which raises RuntimeError.
         t.start()
+        with self._lock:
+            self._connections[(client_sock, target_sock)] = t
+            self.total_connections += 1
 
     def _forward_loop(self, client_sock, target_sock):
         try:
@@ -178,7 +211,7 @@ def _forward_loop(self, client_sock, target_sock):
             pass
         finally:
             with self._lock:
-                self._connections.discard((client_sock, target_sock))
+                self._connections.pop((client_sock, target_sock), None)
             self._close_pair(client_sock, target_sock)
 
     @staticmethod
@@ -189,6 +222,20 @@ def _close_pair(csock, tsock):
             except Exception:
                 pass
 
+    @staticmethod
+    def _shutdown_pair(csock, tsock):
+        """
+        Best-effort shutdown of both sockets in a pair, to interrupt a
+        thread that may be blocked in select()/recv() on them. Safe to
+        call from a different thread than the one using the sockets
+        (unlike close()); does not release the underlying fds.
+        """
+        for s in (csock, tsock):
+            try:
+                s.shutdown(socket.SHUT_RDWR)
+            except OSError:
+                pass
+
 
 class NLBEmulator:
     """
@@ -227,7 +274,10 @@ def __init__(self, discovery_port=0,
         self._node_proxies = {}
         self._discovery_proxy = None
         self._rr_index = 0
-        self._lock = threading.Lock()
+        # RLock: add_node()/remove_node() serialize against each other (see
+        # below) and each also takes the lock internally via
+        # _add_node_proxy(), so a plain Lock would deadlock on reentry.
+        self._lock = threading.RLock()
         self._running = False
 
     def start(self, node_addresses):
@@ -288,13 +338,21 @@ def stop(self):
         log.info("NLB stopped")
 
     def add_node(self, node_id, addr):
-        self._add_node_proxy(node_id, addr)
+        # Serialize against remove_node() so a new proxy/connection can
+        # never be created while another thread's remove_node() is still
+        # in the middle of tearing one down -- TcpProxy.stop() now blocks
+        # until its forwarder threads have joined (see TcpProxy), so by
+        # the time this lock is available, any just-freed fds are fully
+        # reaped and cannot be reused out from under a stale thread.
+        with self._lock:
+            self._add_node_proxy(node_id, addr)
 
     def remove_node(self, node_id):
         with self._lock:
             proxy = self._node_proxies.pop(node_id, None)
+            if proxy:
+                proxy.stop()
         if proxy:
-            proxy.stop()
             log.info("NLB removed node %d", node_id)
 
     def node_port(self, node_id):