From e11836c09246f3a0f69d55936e6033ce7ac48ff1 Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Thu, 30 Jul 2026 10:09:26 +0300 Subject: [PATCH] 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. - NLBEmulator._live_addresses() (read by rr_handler(), the discovery port's round-robin accept handler, from the discovery TcpProxy's own accept-loop thread) now also snapshots self._node_proxies under the same _lock that add_node()/remove_node() mutate it under. Previously it iterated the dict unlocked, so a concurrent add_node()/remove_node() could raise "RuntimeError: dictionary changed size during iteration" on that thread. A follow-up pass over the same synchronization path (Copilot automated review on the PR, flagged low-confidence so not posted as formal review threads, but both genuine) found one more real gap and a missing test: - _shutdown_and_join_connections() joined every forwarder thread with a 5s timeout, then unconditionally popped *every* connection out of self._connections regardless of whether its thread had actually exited. A thread that didn't finish in time was dropped from tracking anyway, so active_connections under-reported live connections, and a later stop()/drop_connections() could never retry shutting it down -- permanently leaking that thread and its fds. Fixed by only popping entries whose thread is confirmed dead (`not thread.is_alive()`) after the join; still-alive entries stay tracked until _forward_loop's own self-removal (already lock-protected and idempotent) reaps them, so a subsequent shutdown call can retry. - Added tests/unit/test_tcp_proxy.py, a checked-in deterministic regression test (TcpProxy has no CCM/cluster dependency, only sockets, so it runs as a plain fast unit test against a local dummy TCP echo backend). It covers: (a) the exact regression above -- neutering _shutdown_pair and shrinking one thread's join wait to deterministically force the "still alive after the timeout" path, and asserting the connection stays tracked until a retried drop_connections() actually reaps it -- and (b) a concurrent stress test that hammers drop_connections() from multiple threads while other threads continuously open/close real connections, asserting no unhandled exceptions and no forwarder threads left alive once stop() returns. 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. - Standalone stress harness driving concurrent readers directly exercising _live_addresses() against concurrent add_node()/ remove_node() churn: pre-fix, 313 "dictionary changed size during iteration" RuntimeErrors over 447k calls in 5s; post-fix, zero errors over 1.9M+ calls across three separate runs. - 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. - New tests/unit/test_tcp_proxy.py: 30/30 clean runs with the active_connections fix in place; with the fix reverted, the targeted regression test failed deterministically 15/15 runs (active_connections incorrectly reported 0 instead of 1 for a still-alive forwarder thread), confirming the test actually catches the bug it targets. - Full tests/unit/ suite: 722 passed, 88 skipped, 0 failed. Fixes #948. Co-Authored-By: Claude Sonnet 5 --- .../standard/test_client_routes.py | 113 ++++++-- tests/unit/test_tcp_proxy.py | 259 ++++++++++++++++++ 2 files changed, 351 insertions(+), 21 deletions(-) create mode 100644 tests/unit/test_tcp_proxy.py diff --git a/tests/integration/standard/test_client_routes.py b/tests/integration/standard/test_client_routes.py index 292eabca30..8e45cf7d93 100644 --- a/tests/integration/standard/test_client_routes.py +++ b/tests/integration/standard/test_client_routes.py @@ -73,7 +73,7 @@ def __init__(self, listen_host, listen_port, target_host, target_port): self._running = False self._thread = None self._lock = threading.Lock() - self._connections = set() + self._connections = {} # (client_sock, target_sock) -> forwarder thread self.total_connections = 0 def start(self): @@ -92,16 +92,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 +116,45 @@ 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): + """ + Shut down (not close) each connection's sockets to unblock its + forwarder thread, then join it. Only the forwarder thread itself + closes its sockets, avoiding a close-vs-still-in-use fd-reuse race. + + stopping=True (stop() only) flips _running to False under the same + lock as the connections snapshot, so no connection registered by + _handle_new_connection can be missed. + """ + with self._lock: + if stopping: + self._running = False + connections = list(self._connections.items()) + for (csock, tsock), _thread in connections: + self._shutdown_pair(csock, tsock) + finished_keys = [] + for (csock, tsock), thread in connections: + thread.join(timeout=5) + if thread.is_alive(): + # Do NOT drop this entry from self._connections: it is + # still a live thread owning open fds. Leaving it tracked + # lets active_connections reflect reality and lets a + # subsequent stop()/drop_connections() retry the shutdown + # and join. _forward_loop() removes its own entry (under + # _lock) once it actually exits, so there's no leak here. + 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) + else: + finished_keys.append((csock, tsock)) + with self._lock: + for key in finished_keys: + self._connections.pop(key, None) + def _run(self): while self._running: try: @@ -153,14 +182,32 @@ 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 then start() atomically under _lock, in that order: + # otherwise a short-lived thread could finish (and clean up) + # before being registered, leaking the entry, or run unseen by + # a concurrent stop()/drop_connections(). Also re-check + # _running, to reject connections after shutdown has begun. + 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: + # Undo registration: join()-ing an unstarted thread later + # would raise RuntimeError. + 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 +225,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 +236,15 @@ def _close_pair(csock, tsock): except Exception: pass + @staticmethod + def _shutdown_pair(csock, tsock): + """Best-effort shutdown (not close) to interrupt a thread blocked in select()/recv().""" + for s in (csock, tsock): + try: + s.shutdown(socket.SHUT_RDWR) + except OSError: + pass + class NLBEmulator: """ @@ -227,7 +283,8 @@ def __init__(self, discovery_port=0, self._node_proxies = {} self._discovery_proxy = None self._rr_index = 0 - self._lock = threading.Lock() + # RLock: add_node() holds it while _add_node_proxy() re-acquires it. + self._lock = threading.RLock() self._running = False def start(self, node_addresses): @@ -288,13 +345,17 @@ 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(): TcpProxy.stop() blocks until + # joined, so by the time we get the lock any freed fds are reaped. + 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): @@ -328,8 +389,18 @@ def _add_node_proxy(self, node_id, addr): node_id, self.LISTEN_HOST, port, addr, self.native_port) def _live_addresses(self): - """IPs of nodes with active proxies.""" - return [p.target_host for p in self._node_proxies.values()] + """ + IPs of nodes with active proxies. + + Snapshots under _lock: rr_handler() (the discovery port's accept + handler) calls this from the discovery TcpProxy's own accept-loop + thread, concurrently with add_node()/remove_node() mutating + _node_proxies from other threads. Without the lock, a node being + added/removed mid-iteration can raise "RuntimeError: dictionary + changed size during iteration". + """ + with self._lock: + return [p.target_host for p in self._node_proxies.values()] def post_client_routes(contact_point, routes): """ diff --git a/tests/unit/test_tcp_proxy.py b/tests/unit/test_tcp_proxy.py new file mode 100644 index 0000000000..71d47d6150 --- /dev/null +++ b/tests/unit/test_tcp_proxy.py @@ -0,0 +1,259 @@ +# Copyright 2026 ScyllaDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# 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. + +""" +Regression tests for the ``TcpProxy`` test helper's connection +shutdown/join synchronization path (GitHub issue #948). + +``TcpProxy`` is defined in +``tests/integration/standard/test_client_routes.py`` because it backs the +Client Routes / NLB integration tests, but it is a plain socket-based +helper with no dependency on a running Cassandra/Scylla cluster or CCM. +These tests exercise it directly against a local dummy TCP echo backend, +so they run as fast, deterministic, checked-in unit tests instead of only +being covered incidentally (and non-deterministically) by the integration +suite. + +Importing that module pulls in ``tests.integration``, whose module-level +code parses ``CASSANDRA_VERSION``/``SCYLLA_VERSION`` into a +``packaging.version.Version`` and raises if neither is set. That parsing +is the only thing gating the import -- no CCM/cluster is started merely by +importing the module -- so a harmless default is provided below when +running standalone (e.g. ``pytest tests/unit``), without overriding a real +value if one is already set (e.g. under the integration test runner). +""" + +import os +import socket +import threading +import time +import unittest +from unittest.mock import patch + +os.environ.setdefault("CASSANDRA_VERSION", "4.0.0") + +from tests.integration.standard.test_client_routes import TcpProxy # noqa: E402 + + +class _EchoServer: + """Minimal threaded TCP echo server used as TcpProxy's backend target.""" + + def __init__(self): + self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self._sock.bind(("127.0.0.1", 0)) + self.port = self._sock.getsockname()[1] + self._sock.listen(128) + self._sock.settimeout(0.2) + self._running = True + self._accept_thread = threading.Thread(target=self._accept_loop, daemon=True) + self._accept_thread.start() + + def _accept_loop(self): + while self._running: + try: + conn, _ = self._sock.accept() + except socket.timeout: + continue + except OSError: + return + threading.Thread(target=self._echo, args=(conn,), daemon=True).start() + + @staticmethod + def _echo(conn): + try: + while True: + data = conn.recv(4096) + if not data: + return + conn.sendall(data) + except OSError: + pass + finally: + try: + conn.close() + except OSError: + pass + + def stop(self): + self._running = False + try: + self._sock.close() + except OSError: + pass + self._accept_thread.join(timeout=2) + + +def _open_client(host, port, timeout=5): + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(timeout) + s.connect((host, port)) + return s + + +class TestTcpProxyShutdownJoin(unittest.TestCase): + """ + Regression coverage for the forwarder-thread bookkeeping bug described + in issue #948: ``_shutdown_and_join_connections`` used to unconditionally + discard every tracked connection from ``_connections``, even ones whose + forwarder thread was still alive after ``thread.join(timeout=5)`` timed + out. That made ``active_connections`` under-report live connections and + made it impossible for a later ``stop()``/``drop_connections()`` call to + retry reaping an orphaned thread, permanently leaking the thread and its + file descriptors. + """ + + def setUp(self): + self.echo = _EchoServer() + self.addCleanup(self.echo.stop) + self.proxy = TcpProxy("127.0.0.1", 0, "127.0.0.1", self.echo.port) + self.proxy.start() + self.addCleanup(self._safe_stop_proxy) + + def _safe_stop_proxy(self): + try: + self.proxy.stop() + except Exception: + pass + + def test_timed_out_forwarder_thread_is_retained_until_it_exits(self): + """ + Exact regression test for the fix: if a forwarder thread does not + exit within the join timeout, its entry must NOT be dropped from + ``_connections`` -- it must stay tracked (so ``active_connections`` + reflects reality and a later shutdown call can retry) until the + thread actually finishes. + """ + client = _open_client(self.proxy.listen_host, self.proxy.listen_port) + self.addCleanup(client.close) + client.sendall(b"ping") + self.assertEqual(client.recv(16), b"ping") + + self.assertEqual(self.proxy.active_connections, 1) + (csock, tsock), thread = list(self.proxy._connections.items())[0] + + # Shrink this thread's effective join timeout so the test doesn't + # have to block for the real 5s timeout, while neutering + # _shutdown_pair so the forwarder genuinely cannot be unblocked -- + # deterministically reproducing "still alive after the timeout". + real_join = thread.join + thread.join = lambda timeout=None: real_join(timeout=0.05) + try: + with patch.object(TcpProxy, "_shutdown_pair", + new=staticmethod(lambda a, b: None)): + self.proxy.drop_connections() + finally: + thread.join = real_join + + # The forwarder thread is still alive: the fixed code must keep + # tracking it instead of discarding the entry. + self.assertTrue(thread.is_alive(), + "test setup issue: forwarder thread should still " + "be running at this point") + self.assertEqual( + self.proxy.active_connections, 1, + "a still-alive forwarder thread's connection entry must not be " + "dropped after its join times out") + self.assertIn((csock, tsock), self.proxy._connections) + + # Retry for real: this time _shutdown_pair actually runs and + # unblocks the thread, so the retry can finish reaping it. + self.proxy.drop_connections() + + thread.join(timeout=5) + self.assertFalse(thread.is_alive()) + self.assertEqual(self.proxy.active_connections, 0) + self.assertNotIn((csock, tsock), self.proxy._connections) + + def test_concurrent_stop_and_drop_leaves_no_live_forwarders(self): + """ + Deterministic stress regression test: concurrently open/close real + connections through the proxy while other threads hammer + drop_connections(), then stop(); assert that (a) no unhandled + exception escaped any thread and (b) no forwarder thread is left + alive or tracked once stop() returns. + """ + errors = [] + stop_event = threading.Event() + forwarder_threads = set() + threads_lock = threading.Lock() + + def client_worker(): + while not stop_event.is_set(): + try: + s = _open_client(self.proxy.listen_host, + self.proxy.listen_port, timeout=1) + except OSError: + continue + try: + with self.proxy._lock: + with threads_lock: + forwarder_threads.update(self.proxy._connections.values()) + s.sendall(b"x") + s.recv(16) + except OSError: + pass + finally: + try: + s.close() + except OSError: + pass + time.sleep(0.005) + + def dropper_worker(): + while not stop_event.is_set(): + try: + self.proxy.drop_connections() + except Exception as e: + errors.append(e) + time.sleep(0.01) + + def thread_excepthook(args): + errors.append(args.exc_value) + + old_hook = threading.excepthook + threading.excepthook = thread_excepthook + try: + client_threads = [threading.Thread(target=client_worker) + for _ in range(4)] + dropper_threads = [threading.Thread(target=dropper_worker) + for _ in range(2)] + for t in client_threads + dropper_threads: + t.start() + + time.sleep(1.0) + + stop_event.set() + for t in client_threads + dropper_threads: + t.join(timeout=5) + self.assertFalse(t.is_alive()) + + self.proxy.stop() + finally: + threading.excepthook = old_hook + + self.assertEqual(errors, [], + "unhandled exceptions during concurrent " + "stop/drop: %r" % (errors,)) + self.assertEqual(self.proxy.active_connections, 0) + + with threads_lock: + collected = list(forwarder_threads) + for t in collected: + self.assertFalse(t.is_alive(), + "%s left alive after stop()" % t.name) + + +if __name__ == "__main__": + unittest.main()