From 07822f57a11a2ea479287950a97967d27742d4d0 Mon Sep 17 00:00:00 2001 From: Greg Scallan Date: Wed, 14 Oct 2026 14:31:49 -0400 Subject: [PATCH 1/9] Raise request attempt bound from 10 to 100 Ten was picked arbitrarily when the retry loop replaced recursion. Memory is flat per attempt once frames are gone, so the bound costs nothing and should be set for time, not memory. Ten sat below the attempts a region reconnect needs, turning recoverable broken pipes into failed writes in production. Co-Authored-By: Claude Opus 5 --- pybase/client.py | 11 ++++++---- tests/test_retry_bounds.py | 41 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 4 deletions(-) diff --git a/pybase/client.py b/pybase/client.py index 9961ff6..d1f7059 100644 --- a/pybase/client.py +++ b/pybase/client.py @@ -40,10 +40,13 @@ logger = logging.getLogger(__name__) -# Retries are bounded because these requests used to retry by recursing: each -# failed attempt kept its frame, row values and built request alive, so a row -# that never succeeds exhausted memory long before Python's recursion limit. -MAX_REQUEST_ATTEMPTS = 10 +# The loop below rebinds each attempt's request, so retrying costs no memory and +# this bound exists only to stop a permanently failing row spinning forever. Keep +# it well above the handful of attempts a region reconnect needs: _handle_exception +# reestablishes the client without sleeping, so a connection broken mid-write is +# recovered by retrying, and too low a bound turns a recoverable broken pipe into +# a failed write. +MAX_REQUEST_ATTEMPTS = 100 class MainClient(object): diff --git a/tests/test_retry_bounds.py b/tests/test_retry_bounds.py index 07e2044..0707681 100644 --- a/tests/test_retry_bounds.py +++ b/tests/test_retry_bounds.py @@ -101,6 +101,47 @@ def fails_once(table, key): finally: gc.enable() + def test_retrying_does_not_accumulate_memory(self): + """Attempts must cost no memory, so the bound can be set for time alone. + + The recursive implementation kept every attempt's request alive at once, + which is what exhausted the container. A loop frees each attempt as the + next one begins, so raising the bound to survive a reconnect is free. + """ + built = [] + + class FakeRequest(object): + def __init__(self): + self.payload = bytearray(1_000_000) + + def rq_type(region, key, values): + request = FakeRequest() + built.append(weakref.ref(request)) + return request + + class FakeRegionClient(object): + def _send_request(self, rq): + raise RecoverableException('broken pipe') + + class FakeRegion(object): + region_client = FakeRegionClient() + + self.client._find_hosting_region = lambda table, key: FakeRegion() + + gc.disable() + try: + with self.assertRaises(RecoverableException): + self.client._mutate('table', 'key', {'cf': {}}, rq_type) + self.assertEqual(len(built), MAX_REQUEST_ATTEMPTS) + self.assertEqual([r for r in built if r() is not None], []) + finally: + gc.enable() + + def test_bound_leaves_room_for_a_region_reconnect(self): + """A broken connection is recovered by retrying, so the bound must exceed + the attempts a reconnect takes. Ten was too low and failed live writes.""" + self.assertGreaterEqual(MAX_REQUEST_ATTEMPTS, 50) + if __name__ == '__main__': unittest.main() From d1bc2a125489ab145c220d36f7c01038c7b5f0e5 Mon Sep 17 00:00:00 2001 From: Greg Scallan Date: Wed, 14 Oct 2026 14:40:01 -0400 Subject: [PATCH 2/9] Cut comment volume in the retry paths Six lines of prose on the bound and five-line docstrings per test restated what names and code already say. Keep one line of non-obvious why each, and drop the step-by-step narration in get() that tracked the code line for line. Co-Authored-By: Claude Opus 5 --- pybase/client.py | 30 +++++++----------------------- tests/test_retry_bounds.py | 24 ++++-------------------- 2 files changed, 11 insertions(+), 43 deletions(-) diff --git a/pybase/client.py b/pybase/client.py index d1f7059..380458f 100644 --- a/pybase/client.py +++ b/pybase/client.py @@ -40,12 +40,8 @@ logger = logging.getLogger(__name__) -# The loop below rebinds each attempt's request, so retrying costs no memory and -# this bound exists only to stop a permanently failing row spinning forever. Keep -# it well above the handful of attempts a region reconnect needs: _handle_exception -# reestablishes the client without sleeping, so a connection broken mid-write is -# recovered by retrying, and too low a bound turns a recoverable broken pipe into -# a failed write. +# Stops a permanently failing row spinning; not a memory limit, since the loop +# frees each attempt. Must exceed a region reconnect's attempts: 10 did not. MAX_REQUEST_ATTEMPTS = 100 @@ -170,27 +166,16 @@ def get(self, table, key, families={}, filters=None): """ for attempt in range(MAX_REQUEST_ATTEMPTS): try: - # Step 0. Set dest_region to None so if an exception is - # thrown in _find_hosting_region, the exception handling - # doesn't break trying to reference dest_region. + # Bound before the call so the except block can read it even when + # _find_hosting_region is what threw. dest_region = None - # Step 1. Figure out where to send it. dest_region = self._find_hosting_region(table, key) - # Step 2. Build the appropriate pb message. rq = request.get_request(dest_region, key, families, filters) - # Step 3. Send the message and twiddle our thumbs. response = dest_region.region_client._send_request(rq) - # Step 4. Success. return Result(response) except PyBaseException as e: - # Step X. Houston, we have an error. The cool thing about how - # this is coded is that exceptions know how to handle themselves. - # All we need to do is call _handle_exception and everything should - # be happy! If it cannot handle itself (unrecoverable) then it will - # re-raise the exception in the handle method and we'll die too. - # - # We pass dest_region in because the handling code needs to know - # which region or region_client it needs to reestablish. + # Reestablishes the region client so the next attempt can succeed, + # or re-raises here if the error is unrecoverable. e._handle_exception(self, dest_region=dest_region) if attempt == MAX_REQUEST_ATTEMPTS - 1: raise @@ -261,8 +246,7 @@ def increment(self, table, key, values): return self._mutate(table, key, values, request.increment_request) def _mutate(self, table, key, values, rq_type): - # Same exact methodology as 'get'. Because all mutate requests have - # equivalent code I've combined them into a single function. + # Shared by put/delete/append/increment; same retry shape as 'get'. for attempt in range(MAX_REQUEST_ATTEMPTS): try: dest_region = None diff --git a/tests/test_retry_bounds.py b/tests/test_retry_bounds.py index 0707681..bb6887b 100644 --- a/tests/test_retry_bounds.py +++ b/tests/test_retry_bounds.py @@ -16,12 +16,7 @@ def _handle_exception(self, main_client, **kwargs): class TestRequestRetryBounds(unittest.TestCase): - """Requests must retry a bounded number of times rather than recursing. - - The recursive implementation kept every attempt's frame, row values and - built request alive, so a row that never succeeded exhausted container - memory before Python's recursion limit was reached. - """ + """Retries must stay bounded and must not accumulate memory per attempt.""" def setUp(self): self.client = MainClient.__new__(MainClient) @@ -64,12 +59,7 @@ def fails_once(table, key): self.assertEqual(len(calls), 2) def test_successful_retry_does_not_retain_payload(self): - """A saved exception would keep the row values alive through its traceback. - - The traceback references the frame that holds the values, so storing the - exception in a local builds a cycle that only the cyclic collector can - break. Re-raising in place keeps the retry path free of that reference. - """ + """A saved exception retains the values via its traceback's frame.""" class Sentinel(object): pass @@ -102,12 +92,7 @@ def fails_once(table, key): gc.enable() def test_retrying_does_not_accumulate_memory(self): - """Attempts must cost no memory, so the bound can be set for time alone. - - The recursive implementation kept every attempt's request alive at once, - which is what exhausted the container. A loop frees each attempt as the - next one begins, so raising the bound to survive a reconnect is free. - """ + """Flat memory per attempt is what makes the bound safe to raise.""" built = [] class FakeRequest(object): @@ -138,8 +123,7 @@ class FakeRegion(object): gc.enable() def test_bound_leaves_room_for_a_region_reconnect(self): - """A broken connection is recovered by retrying, so the bound must exceed - the attempts a reconnect takes. Ten was too low and failed live writes.""" + """10 sat below a reconnect's attempts and failed live writes.""" self.assertGreaterEqual(MAX_REQUEST_ATTEMPTS, 50) From 5a2a3cf2232401925b0482dbbe7cc1dd7573dd9b Mon Sep 17 00:00:00 2001 From: Greg Scallan Date: Wed, 14 Oct 2026 14:49:06 -0400 Subject: [PATCH 3/9] Release the Future so a failed request is not retained A failed Future holds its exception, whose traceback references the sender frame holding the Future and the request, a cycle only the collector breaks. Measured 20 of 20 requests retained with gc off; dropping the local gives 0. Also drops a 3.6-only numeric literal, since setup.py still claims 2.7. Co-Authored-By: Claude Opus 5 --- pybase/region/client.py | 8 +++++- tests/test_retry_bounds.py | 54 ++++++++++++++++++++++++++++++++++++-- 2 files changed, 59 insertions(+), 3 deletions(-) diff --git a/pybase/region/client.py b/pybase/region/client.py index 608a96f..faee1de 100644 --- a/pybase/region/client.py +++ b/pybase/region/client.py @@ -131,7 +131,13 @@ def _send_request(self, rq, lock_timeout=10, _async=False): future = self.thread_pool.submit(self.send_and_receive_rpc, my_id, rq, to_send) if _async: return future - return future.result(timeout=self.thread_pool_timeout) + try: + return future.result(timeout=self.thread_pool_timeout) + finally: + # A failed Future holds its exception, whose traceback references this + # frame, which references the Future. Dropping the local breaks that + # cycle so a retried request is freed without waiting for cyclic gc. + del future # Sending an RPC, listens for the response and builds the correct pbResponse object. # diff --git a/tests/test_retry_bounds.py b/tests/test_retry_bounds.py index bb6887b..c3f9748 100644 --- a/tests/test_retry_bounds.py +++ b/tests/test_retry_bounds.py @@ -3,6 +3,8 @@ import gc import unittest import weakref +from concurrent.futures import ThreadPoolExecutor +from threading import Lock from pybase.client import MAX_REQUEST_ATTEMPTS, MainClient from pybase.exceptions import PyBaseException @@ -92,12 +94,12 @@ def fails_once(table, key): gc.enable() def test_retrying_does_not_accumulate_memory(self): - """Flat memory per attempt is what makes the bound safe to raise.""" + """The loop itself holds no attempt; see the sender test for the Future.""" built = [] class FakeRequest(object): def __init__(self): - self.payload = bytearray(1_000_000) + self.payload = bytearray(1000000) def rq_type(region, key, values): request = FakeRequest() @@ -127,5 +129,53 @@ def test_bound_leaves_room_for_a_region_reconnect(self): self.assertGreaterEqual(MAX_REQUEST_ATTEMPTS, 50) +class TestSendRequestReleasesFuture(unittest.TestCase): + """A failed Future holds its request, so retries must not wait on cyclic gc.""" + + def test_failed_request_is_released_without_cyclic_gc(self): + from pybase.region.client import Client + + client = Client.__new__(Client) + client.call_lock = Lock() + client.call_id = 0 + client.thread_pool = ThreadPoolExecutor(max_workers=2) + client.thread_pool_timeout = 5 + client.host, client.port = b'host', b'1' + + def always_fails(call_id, rq, to_send): + raise RecoverableException('broken pipe') + + client.send_and_receive_rpc = always_fails + + class FakePb(object): + def SerializeToString(self): + return b'x' * 100000 + + class FakeRequest(object): + type = 'Get' + pb = FakePb() + + refs = [] + + def one_attempt(): + # The request must stay local to a frame that only the traceback can + # reach, as it is in production; deleting it here would hide the leak. + request = FakeRequest() + refs.append(weakref.ref(request)) + try: + client._send_request(request) + except RecoverableException: + pass + + gc.disable() + try: + for _ in range(10): + one_attempt() + retained = len([r for r in refs if r() is not None]) + self.assertEqual(retained, 0, '%s of %s requests retained' % (retained, len(refs))) + finally: + gc.enable() + + if __name__ == '__main__': unittest.main() From 8a48079c15d0d5af42588fd031d42d45712587a9 Mon Sep 17 00:00:00 2001 From: Greg Scallan Date: Wed, 14 Oct 2026 14:52:08 -0400 Subject: [PATCH 4/9] Sample live requests during retries, not after Checking weakrefs once _mutate unwound would pass against the recursion it is meant to rule out, which holds every attempt at peak and frees them all on the way out. Sampling at each build shows the loop keeps at most one live; the recursive shape peaks at the attempt count. Co-Authored-By: Claude Opus 5 --- tests/test_retry_bounds.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/tests/test_retry_bounds.py b/tests/test_retry_bounds.py index c3f9748..c90393a 100644 --- a/tests/test_retry_bounds.py +++ b/tests/test_retry_bounds.py @@ -94,14 +94,18 @@ def fails_once(table, key): gc.enable() def test_retrying_does_not_accumulate_memory(self): - """The loop itself holds no attempt; see the sender test for the Future.""" + """At most one attempt is live at a time; recursion held all of them.""" built = [] + live_at_build = [] class FakeRequest(object): def __init__(self): self.payload = bytearray(1000000) def rq_type(region, key, values): + # Sampled here, not after unwinding: recursion holds every attempt at + # peak and frees them all on the way out, which a post-hoc check misses. + live_at_build.append(len([r for r in built if r() is not None])) request = FakeRequest() built.append(weakref.ref(request)) return request @@ -120,6 +124,8 @@ class FakeRegion(object): with self.assertRaises(RecoverableException): self.client._mutate('table', 'key', {'cf': {}}, rq_type) self.assertEqual(len(built), MAX_REQUEST_ATTEMPTS) + # One previous request may still be bound while the next is built. + self.assertLessEqual(max(live_at_build), 1) self.assertEqual([r for r in built if r() is not None], []) finally: gc.enable() From ff616e8daef4b8c0fc274153f1e61e8b01c8a5b6 Mon Sep 17 00:00:00 2001 From: Greg Scallan Date: Wed, 14 Oct 2026 14:54:23 -0400 Subject: [PATCH 5/9] Join the executor before counting retained requests result() can return before the worker releases its work item, which still holds the request arguments, so the count could see unfinished cleanup and fail on scheduling alone. Shut the pool down with cyclic gc still disabled. Co-Authored-By: Claude Opus 5 --- tests/test_retry_bounds.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/test_retry_bounds.py b/tests/test_retry_bounds.py index c90393a..0bb063a 100644 --- a/tests/test_retry_bounds.py +++ b/tests/test_retry_bounds.py @@ -177,6 +177,9 @@ def one_attempt(): try: for _ in range(10): one_attempt() + # result() can return before the worker drops its work item, which also + # holds the request, so join the pool to measure persistent retention. + client.thread_pool.shutdown(wait=True) retained = len([r for r in refs if r() is not None]) self.assertEqual(retained, 0, '%s of %s requests retained' % (retained, len(refs))) finally: From b8a71c85d9dd911f690f86cd312d7c4e495eeccf Mon Sep 17 00:00:00 2001 From: Greg Scallan Date: Wed, 14 Oct 2026 15:02:22 -0400 Subject: [PATCH 6/9] Normalise region client cache keys so broken clients are purged Clients are cached under the protobuf bytes spelling while Client decodes host and port, so the handler's lookup always missed and a dead client was never evicted or slept on. Production shows 496 broken pipes in an hour and zero purges. Key on one str form everywhere; _purge_client also raised TypeError mixing a str host with a bytes separator. Co-Authored-By: Claude Opus 5 --- pybase/client.py | 5 +-- pybase/exceptions.py | 11 +++--- pybase/helpers/__init__.py | 13 +++++++ tests/test_region_cache_key.py | 63 ++++++++++++++++++++++++++++++++++ 4 files changed, 85 insertions(+), 7 deletions(-) create mode 100644 tests/test_region_cache_key.py diff --git a/pybase/client.py b/pybase/client.py index 380458f..16aef89 100644 --- a/pybase/client.py +++ b/pybase/client.py @@ -35,6 +35,7 @@ ZookeeperException ) from .filters import _to_filter +from .helpers import region_cache_key from .region.region import region_from_cell from .request import request @@ -456,8 +457,8 @@ def _create_new_region(self, response, table): elif cell.qualifier == b"server": # Grab the host, port of the Region Server that this region is # hosted on. - server_loc = cell.value host, port = cell.value.split(b':') + server_loc = region_cache_key(host, port) else: continue # Do we have an existing client for this region server already? @@ -516,7 +517,7 @@ def _purge_client(self, region_client): for reg in region_client.regions: self._delete_from_region_cache(reg.table, reg.start_key) self.reverse_client_cache.pop( - region_client.host + b":" + region_client.port, None) + region_cache_key(region_client.host, region_client.port), None) region_client.close() def _purge_region(self, reg): diff --git a/pybase/exceptions.py b/pybase/exceptions.py index 43d0188..0c84133 100644 --- a/pybase/exceptions.py +++ b/pybase/exceptions.py @@ -23,6 +23,8 @@ from builtins import str +from .helpers import region_cache_key + logger = logging.getLogger(__name__) @@ -75,9 +77,8 @@ def _handle_exception(self, main_client, **kwargs): # region_client not set? Then host/port must have been. Fetch the # client given the host, port if self.region_client is None: - concat = self.host + b":" + self.port self.region_client = main_client.reverse_client_cache.get( - concat, None) + region_cache_key(self.host, self.port), None) # we don't care about secondaries, move on if (self.region_client and self.region_client.secondary) or self.secondary: @@ -90,7 +91,8 @@ def _handle_exception(self, main_client, **kwargs): if self.region_client is not None: # We need to make sure that a different thread hasn't already # reestablished to this region. - loc = self.region_client.host + ":" + self.region_client.port + loc = region_cache_key(self.region_client.host, + self.region_client.port) if loc in main_client.reverse_client_cache: # We're the first in and it's our job to kill the client. # Purge it. @@ -102,8 +104,7 @@ def _handle_exception(self, main_client, **kwargs): main_client._purge_client(self.region_client) # Sleep for an arbitrary amount of time. If this returns # False then we've hit our max retry threshold. Die. - key = self.region_client.host + ":" + self.region_client.port - if not _dynamic_sleep(self, key): + if not _dynamic_sleep(self, loc): raise self finally: # Notify all the other threads to wake up because we've handled the diff --git a/pybase/helpers/__init__.py b/pybase/helpers/__init__.py index c36125f..3246372 100644 --- a/pybase/helpers/__init__.py +++ b/pybase/helpers/__init__.py @@ -1,3 +1,16 @@ from __future__ import absolute_import, print_function from . import varint # noqa + + +def region_cache_key(host, port): + """Key for MainClient.reverse_client_cache, as a str. + + Region servers arrive as bytes from protobuf but Client decodes them, so the + two spellings have to be normalised or a cached client is never found again. + """ + if isinstance(host, bytes): + host = host.decode('utf8') + if isinstance(port, bytes): + port = port.decode('utf8') + return host + ':' + port diff --git a/tests/test_region_cache_key.py b/tests/test_region_cache_key.py new file mode 100644 index 0000000..2de16af --- /dev/null +++ b/tests/test_region_cache_key.py @@ -0,0 +1,63 @@ +from __future__ import absolute_import, print_function, unicode_literals + +import unittest + +from pybase.client import MainClient +from pybase.exceptions import RegionServerException +from pybase.helpers import region_cache_key +from pybase.region.client import Client + + +def _region_client(host, port): + """A Client as pybase builds it: __init__ decodes host and port to str.""" + client = Client.__new__(Client) + client.host = host.decode('utf8') if isinstance(host, bytes) else host + client.port = port.decode('utf8') if isinstance(port, bytes) else port + client.secondary = False + client.regions = [] + client.closed = False + client.close = lambda: setattr(client, 'closed', True) + return client + + +class TestRegionCacheKey(unittest.TestCase): + """Region servers arrive as bytes but Client stores them decoded.""" + + def test_bytes_and_str_spellings_agree(self): + self.assertEqual(region_cache_key(b'rs1', b'16020'), + region_cache_key('rs1', '16020')) + + def test_handler_reaches_purge_for_a_cached_client(self): + """Covers the handler calling purge at all, not the key spelling.""" + client = _region_client(b'rs1.example', b'16020') + main = MainClient.__new__(MainClient) + main.reverse_client_cache = {region_cache_key(b'rs1.example', b'16020'): client} + + purged = [] + main._purge_client = lambda c: purged.append(c) + + error = RegionServerException(region_client=client) + try: + error._handle_exception(main) + except RegionServerException: + pass + + self.assertEqual(purged, [client], + 'cached client was not purged, so a retry reuses it') + + def test_purge_removes_the_entry_the_cache_was_stored_under(self): + """Mixing a str host with a bytes separator raised, so nothing was purged.""" + client = _region_client(b'rs1.example', b'16020') + main = MainClient.__new__(MainClient) + main.reverse_client_cache = {region_cache_key(b'rs1.example', b'16020'): client} + main.region_cache = {} + main._cache_lock = __import__('threading').Lock() + + main._purge_client(client) + + self.assertEqual(main.reverse_client_cache, {}) + self.assertTrue(client.closed) + + +if __name__ == '__main__': + unittest.main() From accc5cabf41a34e9455f428dc2da1492a76e7d99 Mon Sep 17 00:00:00 2001 From: Greg Scallan Date: Wed, 14 Oct 2026 15:14:34 -0400 Subject: [PATCH 7/9] Retire clients cleanly now that eviction actually happens Working eviction exposed three leaks it had been hiding. A late failure from a retired client could evict the healthy replacement sharing its address, so purge only when the cached entry is that client. close() left the executor's workers running, and the bucket key held the client itself. Co-Authored-By: Claude Opus 5 --- pybase/client.py | 9 +++- pybase/exceptions.py | 3 ++ pybase/region/client.py | 4 ++ tests/test_client_retirement.py | 77 +++++++++++++++++++++++++++++++++ 4 files changed, 91 insertions(+), 2 deletions(-) create mode 100644 tests/test_client_retirement.py diff --git a/pybase/client.py b/pybase/client.py index 16aef89..40a36f1 100644 --- a/pybase/client.py +++ b/pybase/client.py @@ -514,10 +514,15 @@ def _purge_client(self, region_client): # our cache, delete the reverse lookup entry and close the client # clearing up any file descriptors. with self._cache_lock: + key = region_cache_key(region_client.host, region_client.port) + if self.reverse_client_cache.get(key) is not region_client: + # A healthy client already took this address, so a late failure + # from the old one must retire it without evicting its successor. + region_client.close() + return for reg in region_client.regions: self._delete_from_region_cache(reg.table, reg.start_key) - self.reverse_client_cache.pop( - region_cache_key(region_client.host, region_client.port), None) + self.reverse_client_cache.pop(key, None) region_client.close() def _purge_region(self, reg): diff --git a/pybase/exceptions.py b/pybase/exceptions.py index 0c84133..e06a1a6 100644 --- a/pybase/exceptions.py +++ b/pybase/exceptions.py @@ -295,6 +295,9 @@ def _let_all_through(exception, data): _buckets_lock.acquire() my_sem = _buckets[my_tuple] my_sem.release() + # The key holds the region client, so a retained bucket outlives every + # retired client. Waiters already hold their own reference to the semaphore. + _buckets.pop(my_tuple, None) _buckets_lock.release() diff --git a/pybase/region/client.py b/pybase/region/client.py index faee1de..080e982 100644 --- a/pybase/region/client.py +++ b/pybase/region/client.py @@ -237,6 +237,10 @@ def close(self): self.shutting_down = True for sock in self.sock_pool: sock.close() + if self.thread_pool is not None: + # Closing sockets leaves the pool's workers running, so a retired + # client would strand its threads. + self.thread_pool.shutdown(wait=False) # We could still have greenlets waiting in the bad_call_id pools! Wake # them up so they can fail to error handling as well. diff --git a/tests/test_client_retirement.py b/tests/test_client_retirement.py new file mode 100644 index 0000000..7dcc4de --- /dev/null +++ b/tests/test_client_retirement.py @@ -0,0 +1,77 @@ +from __future__ import absolute_import, print_function, unicode_literals + +import threading +import unittest +from concurrent.futures import ThreadPoolExecutor + +from pybase import exceptions +from pybase.client import MainClient +from pybase.exceptions import RegionServerException +from pybase.helpers import region_cache_key +from pybase.region.client import Client + + +def _region_client(host, port, with_pool=False): + client = Client.__new__(Client) + client.host = host + client.port = port + client.secondary = False + client.regions = [] + client.shutting_down = False + client.sock_pool = [] + client.thread_pool = ThreadPoolExecutor(max_workers=1) if with_pool else None + return client + + +def _main_with(cached): + main = MainClient.__new__(MainClient) + main.reverse_client_cache = dict(cached) + main.region_cache = {} + main._cache_lock = threading.Lock() + return main + + +class TestClientRetirement(unittest.TestCase): + + def test_late_failure_does_not_evict_the_replacement(self): + """Normalised keys let a retired client's failure match its successor.""" + key = region_cache_key('rs1', '16020') + retired = _region_client('rs1', '16020') + replacement = _region_client('rs1', '16020') + main = _main_with({key: replacement}) + + main._purge_client(retired) + + self.assertIs(main.reverse_client_cache.get(key), replacement) + + def test_purging_the_cached_client_still_evicts_it(self): + key = region_cache_key('rs1', '16020') + cached = _region_client('rs1', '16020') + main = _main_with({key: cached}) + + main._purge_client(cached) + + self.assertEqual(main.reverse_client_cache, {}) + + def test_close_retires_the_executor(self): + """Closing sockets alone leaves the pool's worker threads running.""" + client = _region_client('rs1', '16020', with_pool=True) + pool = client.thread_pool + + client.close() + + self.assertTrue(pool._shutdown) + + def test_repeated_retirement_does_not_grow_the_bucket_map(self): + """The bucket key holds the region client, so entries must be released.""" + exceptions._buckets.clear() + for _ in range(25): + client = _region_client('rs1', '16020') + error = RegionServerException(region_client=client) + exceptions._let_one_through(error, client) + exceptions._let_all_through(error, client) + self.assertEqual(len(exceptions._buckets), 0) + + +if __name__ == '__main__': + unittest.main() From c8f55ca888e3330dd67c302e3afececbe815a2a7 Mon Sep 17 00:00:00 2001 From: Greg Scallan Date: Wed, 14 Oct 2026 15:22:11 -0400 Subject: [PATCH 8/9] Claim retry buckets atomically and retry a retired pool Popping the bucket left a window: a caller could look one up, have it retired, then claim the dead semaphore and later release a different one. Claim under the map lock. Submitting to a shut-down pool also raised RuntimeError, which the retry handlers do not catch, so translate it. Co-Authored-By: Claude Opus 5 --- pybase/exceptions.py | 9 ++++---- pybase/region/client.py | 10 ++++++++- tests/test_client_retirement.py | 38 +++++++++++++++++++++++++++++++++ 3 files changed, 52 insertions(+), 5 deletions(-) diff --git a/pybase/exceptions.py b/pybase/exceptions.py index e06a1a6..de2e675 100644 --- a/pybase/exceptions.py +++ b/pybase/exceptions.py @@ -271,13 +271,14 @@ class MalformedValues(PyBaseException): # _let_all_through. def _let_one_through(exception, data): my_tuple = (exception.__class__.__name__, data) - # Grab my relevant semaphore. + # Claim under the same lock that guards the map, or the bucket can be retired + # between lookup and acquisition and this caller ends up owning a semaphore + # nobody will ever release, stranding everyone waiting on it. _buckets_lock.acquire() my_sem = _buckets[my_tuple] + claimed = my_sem.acquire(blocking=False) _buckets_lock.release() - # Try to non-blocking acquire my semaphore. If I get it, woohoo! Otherwise - # get comfy because we're sitting on the semaphore. - if my_sem.acquire(blocking=False): + if claimed: # Look at me - I'm the captain now. return True else: diff --git a/pybase/region/client.py b/pybase/region/client.py index 080e982..f59cfd0 100644 --- a/pybase/region/client.py +++ b/pybase/region/client.py @@ -128,7 +128,15 @@ def _send_request(self, rq, lock_timeout=10, _async=False): to_send += serialized_header + rpc_length_bytes + serialized_rpc # send and receive the request - future = self.thread_pool.submit(self.send_and_receive_rpc, my_id, rq, to_send) + try: + future = self.thread_pool.submit(self.send_and_receive_rpc, my_id, rq, to_send) + except RuntimeError: + # Retired between the region lookup and this submit; surface it as a + # region failure so the caller retries instead of dying on a + # RuntimeError the retry handlers do not catch. + if self.shutting_down: + raise RegionServerException(region_client=self) + raise if _async: return future try: diff --git a/tests/test_client_retirement.py b/tests/test_client_retirement.py index 7dcc4de..3731141 100644 --- a/tests/test_client_retirement.py +++ b/tests/test_client_retirement.py @@ -73,5 +73,43 @@ def test_repeated_retirement_does_not_grow_the_bucket_map(self): self.assertEqual(len(exceptions._buckets), 0) + def test_bucket_round_trip_claims_and_releases_one_semaphore(self): + """Covers the claim/release round trip. The interleaving that motivated + making the claim atomic is not reproduced here; see the PR discussion.""" + exceptions._buckets.clear() + client = _region_client('rs1', '16020') + first = RegionServerException(region_client=client) + second = RegionServerException(region_client=client) + + self.assertTrue(exceptions._let_one_through(first, client)) + # Owner finishes and retires the bucket. + exceptions._let_all_through(first, client) + # A later caller must claim a live bucket, not the retired semaphore. + self.assertTrue(exceptions._let_one_through(second, client)) + claimed = exceptions._buckets[(second.__class__.__name__, client)] + exceptions._let_all_through(second, client) + self.assertEqual(claimed._value, 1, 'released a different semaphore') + + def test_submit_after_retirement_is_a_region_failure(self): + """RuntimeError from a retired pool would bypass the retry handlers.""" + client = _region_client('rs1', '16020', with_pool=True) + client.call_lock = threading.Lock() + client.call_id = 0 + client.thread_pool_timeout = 5 + + class FakePb(object): + def SerializeToString(self): + return b'x' + + class FakeRequest(object): + type = 'Get' + pb = FakePb() + + client.close() + + with self.assertRaises(RegionServerException): + client._send_request(FakeRequest()) + + if __name__ == '__main__': unittest.main() From d5f9ba3dc9556f7b8d52733d861889fb84f1111e Mon Sep 17 00:00:00 2001 From: Greg Scallan Date: Wed, 14 Oct 2026 15:22:24 -0400 Subject: [PATCH 9/9] Drop a stray blank line Co-Authored-By: Claude Opus 5 --- tests/test_client_retirement.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/test_client_retirement.py b/tests/test_client_retirement.py index 3731141..8e9b744 100644 --- a/tests/test_client_retirement.py +++ b/tests/test_client_retirement.py @@ -72,7 +72,6 @@ def test_repeated_retirement_does_not_grow_the_bucket_map(self): exceptions._let_all_through(error, client) self.assertEqual(len(exceptions._buckets), 0) - def test_bucket_round_trip_claims_and_releases_one_semaphore(self): """Covers the claim/release round trip. The interleaving that motivated making the claim atomic is not reproduced here; see the PR discussion."""