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):
@@ -92,16 +94,12 @@ def start(self):
                  self.target_host, self.target_port)
 
     def stop(self):
-        self._running = False
         if self._server_sock:
             try:
                 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(stopping=True)
         if self._thread:
             self._thread.join(timeout=5)
         log.info("TcpProxy stopped %s:%d", self.listen_host, self.listen_port)
@@ -120,12 +118,61 @@ 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, stopping=False):
+        """
+        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.
+
+        :param stopping: if True (only ``stop()`` passes this), flips
+            ``_running`` to False in the *same* critical section that
+            takes the ``_connections`` snapshot below. This is the other
+            half of the atomic register-then-start protocol in
+            ``_handle_new_connection``: that method only ever publishes a
+            new connection while holding ``_lock`` and after confirming
+            ``_running`` is still True, so once this method has taken
+            _running False and snapshotted _connections under the same
+            lock, no connection can be registered afterwards that this
+            snapshot fails to observe -- there is no window where a new
+            connection slips in after shutdown has effectively begun.
+            ``drop_connections()`` must NOT do this: it is meant to drop
+            in-flight connections while leaving the proxy running (e.g.
+            to exercise client reconnect logic), so it leaves _running
+            untouched and new connections keep being accepted normally.
+        """
+        with self._lock:
+            if stopping:
+                self._running = False
+            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 +200,61 @@ 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)
-        t.start()
+        # Register the thread into _connections and start() it as a
+        # single atomic sequence, both under _lock, registration BEFORE
+        # start(). Doing this any other way reopens a race:
+        #  - register-after-start (the previous approach) leaves a
+        #    window where the thread is running but not yet in
+        #    _connections; a very short-lived connection (e.g.
+        #    immediate EOF) can run _forward_loop's `finally` -- which
+        #    pops (a no-op, since the key isn't there yet) and closes
+        #    the sockets -- before this method inserts the entry,
+        #    leaving a stale/leaked dict entry that nothing ever cleans
+        #    up. Worse, a concurrent stop()/drop_connections() can
+        #    snapshot _connections during that same window, miss this
+        #    thread entirely, and (for stop()) return as if fully shut
+        #    down while the thread is still actively using its sockets.
+        #  - start-after-register-but-not-atomically (i.e. two separate
+        #    critical sections) reopens the same window between the two
+        #    lock acquisitions.
+        # Keeping both operations inside one critical section closes
+        # the window: _shutdown_and_join_connections() also takes its
+        # _connections snapshot under _lock (see there), so it can only
+        # ever observe this connection in a state where start() has
+        # already been called -- eliminating both the leaked-entry case
+        # and any join-before-start RuntimeError risk.
+        #
+        # We also re-check _running here, under the very same lock that
+        # stop() uses to flip it False (see _shutdown_and_join_connections):
+        # if shutdown has already begun, reject the new connection
+        # outright instead of letting it slip in afterwards.
+        with self._lock:
+            if not self._running:
+                target_sock.close()
+                client_sock.close()
+                return
+            self._connections[(client_sock, target_sock)] = t
+            self.total_connections += 1
+            try:
+                t.start()
+            except Exception as e:
+                # If start() itself fails (e.g. the OS refuses to create a
+                # new thread), undo the registration immediately -- an
+                # entry left behind here would point at a thread that was
+                # never started, and a later stop()/drop_connections()
+                # calling .join() on it would hit the exact
+                # "cannot join thread before it is started" RuntimeError
+                # this whole atomic register+start sequence exists to
+                # prevent.
+                self._connections.pop((client_sock, target_sock), None)
+                self.total_connections -= 1
+                log.warning("TcpProxy %s:%d failed to start forwarder thread: %s",
+                            self.listen_host, self.listen_port, e)
+                client_sock.close()
+                target_sock.close()
 
     def _forward_loop(self, client_sock, target_sock):
         try:
@@ -178,7 +272,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 +283,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 +335,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 +399,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):