diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c1783c..941ba64 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,9 @@ +## 4.4.5 - 2026-08-03 + +- Fix `ThreadPoolExecutor` leak on the exception path when `attempt_timeout` is set (#36). The executor created per attempt was only shut down on the timeout path, leaking worker threads whenever the target raised an exception. `_retry_loop_sync` now wraps the executor in `try/finally` so `_executor.shutdown(wait=False)` runs on every code path (success, timeout, and exception). +- Fix handler `details` dict always containing empty `args` and `kwargs` (#73). The real call arguments were never threaded into `RetryState`/`RetryCallState`, so handlers reading `details["args"]`/`details["kwargs"]` (e.g. `details["kwargs"]["query_id"]`) got `()`/`{}` and raised `KeyError`. `args`/`kwargs` are now plumbed from every entry point (`on_exception`/`on_predicate` sync+async, sync/async generators, `Retrying.call`/`async_call`, `RetryingCaller`/`AsyncRetryingCaller`, `hedge`) into the retry loop, and `RetryState`/`RetryCallState` are constructed with them so every handler (`on_attempt`, `on_backoff`, `on_giveup`, `on_success`, `before`, `after`, `before_sleep`) receives the actual call arguments. +- Add regression tests for both issues: executor thread leak (decorator, functional API, `RetryingCaller`, success path) and handler `details` args/kwargs (sync+async, generators, `Retrying`/`RetryingCaller`/`AsyncRetryingCaller`, `on_backoff`/`on_giveup`/`on_success`/`on_attempt`/`before`/`after`/`before_sleep`), including the `KeyError` reproduction from the issue. + ## 4.4.4 - 2026-07-25 - Fix `hedge()` functional API not accepting arguments for the target function (#43). Added `args` and `kw` keyword-only parameters to `hedge()`, `_hedge_sync()`, and `_hedge_async()`, forwarding them to the target via `_make_hedge_target`. diff --git a/backon/_decorator.py b/backon/_decorator.py index 80056d5..c7456eb 100644 --- a/backon/_decorator.py +++ b/backon/_decorator.py @@ -355,6 +355,8 @@ async def wrapper(*args: P.args, **kwargs: P.kwargs): lambda: _collect_async_gen(target(*args, **kwargs)), wait_gen, sleep=_sleep, + args=args, + kwargs=kwargs, **_kw, ) if collected is not None: @@ -377,6 +379,8 @@ def wrapper(*args: P.args, **kwargs: P.kwargs): lambda: list(target(*args, **kwargs)), wait_gen, sleep=_sleep, + args=args, + kwargs=kwargs, **_kw, ) if collected is not None: @@ -402,6 +406,8 @@ async def wrapped(): wrapped, wait_gen, sleep=_sleep, + args=args, + kwargs=kwargs, **_kw, ), ) @@ -422,6 +428,8 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: lambda: target(*args, **kwargs), wait_gen, sleep=_sleep, + args=args, + kwargs=kwargs, **_kw, ), ) @@ -589,6 +597,8 @@ async def wrapper(*args: P.args, **kwargs: P.kwargs): lambda: _collect_async_gen(target(*args, **kwargs)), wait_gen, sleep=_sleep, + args=args, + kwargs=kwargs, **_kw, ) if collected is not None: @@ -611,6 +621,8 @@ def wrapper(*args: P.args, **kwargs: P.kwargs): lambda: list(target(*args, **kwargs)), wait_gen, sleep=_sleep, + args=args, + kwargs=kwargs, **_kw, ) if collected is not None: @@ -636,6 +648,8 @@ async def wrapped(): wrapped, wait_gen, sleep=_sleep, + args=args, + kwargs=kwargs, **_kw, ), ) @@ -656,6 +670,8 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: lambda: target(*args, **kwargs), wait_gen, sleep=_sleep, + args=args, + kwargs=kwargs, **_kw, ), ) diff --git a/backon/_hedging.py b/backon/_hedging.py index 4d347d4..4593e37 100644 --- a/backon/_hedging.py +++ b/backon/_hedging.py @@ -131,6 +131,8 @@ def _hedge_sync( retry_error_callback=None, raise_on_giveup=True, wait_gen_kwargs=wait_gen_kwargs, + args=args, + kwargs=kw or {}, ) futures.add(fut) @@ -197,6 +199,8 @@ async def run_hedge(): retry_error_callback=None, raise_on_giveup=True, wait_gen_kwargs=wait_gen_kwargs, + args=args, + kwargs=kw or {}, ) tasks = [asyncio.create_task(run_hedge()) for _ in range(max_hedge)] diff --git a/backon/_retry/_api.py b/backon/_retry/_api.py index 822fbee..e7b69c5 100644 --- a/backon/_retry/_api.py +++ b/backon/_retry/_api.py @@ -62,6 +62,8 @@ def _retry_sync( _holder: dict | None = None, rate_limit: RateLimiter | None = None, attempt_timeout: float | None = None, + args: tuple = (), + kwargs: dict | None = None, ) -> Any: if wait_gen_kwargs is None: wait_gen_kwargs = {} @@ -117,6 +119,8 @@ def _retry_sync( _holder=_holder, rate_limit=rate_limit, attempt_timeout=attempt_timeout, + args=args, + kwargs=kwargs, ) @@ -149,6 +153,8 @@ async def _retry_async( _holder: dict | None = None, rate_limit: RateLimiter | None = None, attempt_timeout: float | None = None, + args: tuple = (), + kwargs: dict | None = None, ) -> Any: if wait_gen_kwargs is None: wait_gen_kwargs = {} @@ -204,6 +210,8 @@ async def _retry_async( _holder=_holder, rate_limit=rate_limit, attempt_timeout=attempt_timeout, + args=args, + kwargs=kwargs, ) diff --git a/backon/_retry/_classes.py b/backon/_retry/_classes.py index dda5b64..7c02fc8 100644 --- a/backon/_retry/_classes.py +++ b/backon/_retry/_classes.py @@ -275,6 +275,8 @@ def wrapped(): _holder=_holder, rate_limit=self._rate_limit, attempt_timeout=self._attempt_timeout, + args=args, + kwargs=kwargs, ) finally: self._state = _holder.get("state") @@ -316,6 +318,8 @@ async def wrapped(): _holder=_holder, rate_limit=self._rate_limit, attempt_timeout=self._attempt_timeout, + args=args, + kwargs=kwargs, ) finally: self._state = _holder.get("state") @@ -426,6 +430,8 @@ def __call__(self, target: Callable[..., Any], *args: Any, **kwargs: Any) -> Any after=self._after, rate_limit=self._rate_limit, attempt_timeout=self._attempt_timeout, + args=args, + kwargs=kwargs, ) def copy(self) -> RetryingCaller: @@ -555,6 +561,8 @@ async def wrapped(): after=self._after, rate_limit=self._rate_limit, attempt_timeout=self._attempt_timeout, + args=args, + kwargs=kwargs, ) def copy(self) -> AsyncRetryingCaller: diff --git a/backon/_retry/_fast.py b/backon/_retry/_fast.py index ce2e3f5..58746dc 100644 --- a/backon/_retry/_fast.py +++ b/backon/_retry/_fast.py @@ -281,6 +281,8 @@ def _retry_fast_sync_inner( _holder=None, rate_limit=None, attempt_timeout=None, + args=(), + kwargs=None, ): if not is_enabled(): return target() @@ -346,6 +348,8 @@ def _retry_fast_sync_inner( _holder=_holder, rate_limit=rate_limit, attempt_timeout=attempt_timeout, + args=args, + kwargs=kwargs, ) @@ -372,6 +376,8 @@ async def _retry_fast_async_inner( _holder=None, rate_limit=None, attempt_timeout=None, + args=(), + kwargs=None, ): if not is_enabled(): return await target() @@ -437,4 +443,6 @@ async def _retry_fast_async_inner( _holder=_holder, rate_limit=rate_limit, attempt_timeout=attempt_timeout, + args=args, + kwargs=kwargs, ) diff --git a/backon/_retry/_inner.py b/backon/_retry/_inner.py index b9465d9..1d47bf5 100644 --- a/backon/_retry/_inner.py +++ b/backon/_retry/_inner.py @@ -31,6 +31,8 @@ def _retry_sync_inner( _holder=None, rate_limit=None, attempt_timeout=None, + args=(), + kwargs=None, ): if not is_enabled(): return target() @@ -64,6 +66,8 @@ def _retry_sync_inner( _holder=_holder, rate_limit=rate_limit, attempt_timeout=attempt_timeout, + args=args, + kwargs=kwargs, ) @@ -90,6 +94,8 @@ async def _retry_async_inner( _holder=None, rate_limit=None, attempt_timeout=None, + args=(), + kwargs=None, ): if not is_enabled(): return await target() @@ -123,4 +129,6 @@ async def _retry_async_inner( _holder=_holder, rate_limit=rate_limit, attempt_timeout=attempt_timeout, + args=args, + kwargs=kwargs, ) diff --git a/backon/_retry/_loops.py b/backon/_retry/_loops.py index ea99188..8c20b11 100644 --- a/backon/_retry/_loops.py +++ b/backon/_retry/_loops.py @@ -42,11 +42,15 @@ def _retry_loop_sync( _holder=None, rate_limit=None, attempt_timeout=None, + args=(), + kwargs=None, ): - state = RetryState(target=target) + state = RetryState(target=target, args=args, kwargs=kwargs or {}) start_time = _now() state.start_time = start_time - call_state = RetryCallState(fn=target, start_time=start_time) + call_state = RetryCallState( + fn=target, start_time=start_time, args=args, kwargs=kwargs or {} + ) if _holder is not None: _holder["state"] = state _holder["call_state"] = call_state @@ -70,14 +74,15 @@ def _retry_loop_sync( try: if attempt_timeout is not None: _executor = ThreadPoolExecutor(max_workers=1) - _fut = _executor.submit(target) try: - ret = _fut.result(timeout=attempt_timeout) - except _FuturesTimeoutError: - _fut.cancel() + _fut = _executor.submit(target) + try: + ret = _fut.result(timeout=attempt_timeout) + except _FuturesTimeoutError: + _fut.cancel() + raise AttemptTimeoutError() from None + finally: _executor.shutdown(wait=False) - raise AttemptTimeoutError() from None - _executor.shutdown(wait=False) else: ret = target() except TryAgain: @@ -192,11 +197,15 @@ async def _retry_loop_async( _holder=None, rate_limit=None, attempt_timeout=None, + args=(), + kwargs=None, ): - state = RetryState(target=target) + state = RetryState(target=target, args=args, kwargs=kwargs or {}) start_time = _now() state.start_time = start_time - call_state = RetryCallState(fn=target, start_time=start_time) + call_state = RetryCallState( + fn=target, start_time=start_time, args=args, kwargs=kwargs or {} + ) if _holder is not None: _holder["state"] = state _holder["call_state"] = call_state diff --git a/backon/_typing.py b/backon/_typing.py index 59b840f..3d224dc 100644 --- a/backon/_typing.py +++ b/backon/_typing.py @@ -1,15 +1,14 @@ from __future__ import annotations import logging -import sys from collections.abc import Callable, Coroutine, Sequence from typing import Any, TypedDict, TypeVar, Union from backon._wait_gen import _Wait -if sys.version_info >= (3, 10): +try: from typing import ParamSpec -else: +except ImportError: # pragma: no cover class ParamSpec: # type: ignore[no-redef] # pragma: no cover def __init__(self, name: str) -> None: ... diff --git a/pyproject.toml b/pyproject.toml index c317554..830e094 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "pdm.backend" [project] name = "backon" -version = "4.4.4" +version = "4.4.5" description = "Function decoration for backoff and retry" readme = "README.md" license = "MIT" @@ -81,7 +81,7 @@ testpaths = ["tests"] asyncio_mode = "auto" [tool.mypy] -python_version = "3.10" +python_version = "3.13" strict = false ignore_missing_imports = true warn_unused_ignores = false diff --git a/tests/test_coverage_v4.py b/tests/test_coverage_v4.py index 2075f8d..86453bc 100644 --- a/tests/test_coverage_v4.py +++ b/tests/test_coverage_v4.py @@ -533,6 +533,94 @@ def fn(): assert handler_calls == [1, 2, 1, 2] +class TestOnPredicateAsyncGeneratorEmpty: + @pytest.mark.asyncio + async def test_async_generator_empty(self): + calls = [] + + @backon.on_predicate( + backon.constant, interval=0, jitter=None, max_tries=3, raise_on_giveup=False + ) + async def gen(): + calls.append(1) + if False: + yield + + result = [item async for item in gen()] + assert result == [] + assert len(calls) == 3 + + +class TestOnPredicateAsyncGeneratorDisabledEmpty: + @pytest.mark.asyncio + async def test_async_generator_empty_disabled(self): + was = backon._common.is_enabled() + backon.disable() + try: + calls = [] + + @backon.on_predicate( + backon.constant, + interval=0, + jitter=None, + max_tries=3, + raise_on_giveup=False, + ) + async def gen(): + calls.append(1) + if False: + yield + + result = [item async for item in gen()] + assert result == [] + assert len(calls) == 1 + finally: + if was: + backon.enable() + + +class TestOnExceptionAsyncGeneratorEmpty: + @pytest.mark.asyncio + async def test_async_generator_empty(self): + calls = [] + + @backon.on_exception( + backon.constant, ValueError, interval=0, jitter=None, max_tries=3 + ) + async def gen(): + calls.append(1) + if False: + yield + + result = [item async for item in gen()] + assert result == [] + assert len(calls) == 1 + + +class TestOnExceptionAsyncGeneratorDisabledEmpty: + @pytest.mark.asyncio + async def test_async_generator_empty_disabled(self): + was = backon._common.is_enabled() + backon.disable() + try: + calls = [] + + @backon.on_exception( + backon.constant, ValueError, interval=0, jitter=None, max_tries=3 + ) + async def gen(): + calls.append(1) + if False: + yield + + result = [item async for item in gen()] + assert result == [] + assert len(calls) == 1 + finally: + if was: + backon.enable() + + class TestOnPredicateAsyncGeneratorDisabledPath: """Covers disabled branch for async gen in on_predicate.""" diff --git a/tests/test_issues_36_73.py b/tests/test_issues_36_73.py new file mode 100644 index 0000000..5e083d5 --- /dev/null +++ b/tests/test_issues_36_73.py @@ -0,0 +1,453 @@ +import contextlib +import threading +import time + +import pytest + +import backon + + +def _wait_for_thread_baseline(baseline, timeout=1.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if threading.active_count() <= baseline: + break + time.sleep(0.01) + return threading.active_count() <= baseline + + +class TestIssue36ExecutorLeak: + def test_no_thread_leak_on_exception_path(self): + baseline = threading.active_count() + + @backon.on_exception( + backon.constant, + ValueError, + max_tries=4, + interval=0, + jitter=None, + attempt_timeout=5.0, + ) + def fn(): + raise ValueError("fail") + + with contextlib.suppress(ValueError): + fn() + + assert _wait_for_thread_baseline(baseline) + + def test_no_thread_leak_on_exception_path_functional_api(self): + baseline = threading.active_count() + + def fn(): + raise ValueError("fail") + + with contextlib.suppress(ValueError): + backon.retry( + fn, + backon.constant, + exception=ValueError, + max_tries=4, + interval=0, + jitter=None, + attempt_timeout=5.0, + ) + + assert _wait_for_thread_baseline(baseline) + + def test_no_thread_leak_on_exception_path_retrying_caller(self): + baseline = threading.active_count() + + def fn(): + raise ValueError("fail") + + caller = backon.RetryingCaller( + backon.constant, + max_tries=4, + jitter=None, + interval=0, + attempt_timeout=5.0, + ) + caller = caller.on(ValueError) + with contextlib.suppress(ValueError): + caller(fn) + + assert _wait_for_thread_baseline(baseline) + + def test_no_thread_leak_on_success_path(self): + baseline = threading.active_count() + calls = [] + + @backon.on_exception( + backon.constant, + ValueError, + max_tries=3, + interval=0, + jitter=None, + attempt_timeout=5.0, + ) + def fn(): + calls.append(1) + return "ok" + + assert fn() == "ok" + assert len(calls) == 1 + + assert _wait_for_thread_baseline(baseline) + + +class TestIssue73HandlerDetailsArgs: + def test_on_exception_backoff_receives_kwargs(self): + seen = [] + + def handler(details): + seen.append((details["args"], dict(details["kwargs"]))) + + @backon.on_exception( + backon.constant, + ValueError, + max_tries=3, + interval=0, + jitter=None, + on_backoff=handler, + ) + def do_something(num, *, query_id, flag=True): + raise ValueError("boom") + + with contextlib.suppress(ValueError): + do_something(42, query_id=7, flag=False) + + assert len(seen) == 2 + assert all( + args == (42,) and kw == {"query_id": 7, "flag": False} for args, kw in seen + ) + + def test_on_exception_giveup_and_attempt_receive_args(self): + events = [] + + def on_attempt(details): + events.append(("attempt", details["args"], dict(details["kwargs"]))) + + def on_giveup(details): + events.append(("giveup", details["args"], dict(details["kwargs"]))) + + @backon.on_exception( + backon.constant, + ValueError, + max_tries=2, + interval=0, + jitter=None, + on_attempt=on_attempt, + on_giveup=on_giveup, + ) + def flaky(x, y=0): + raise ValueError("fail") + + with contextlib.suppress(ValueError): + flaky(1, y=2) + + assert ("attempt", (1,), {"y": 2}) in events + assert ("giveup", (1,), {"y": 2}) in events + + def test_on_predicate_success_receives_args(self): + seen = [] + + def on_success(details): + seen.append((details["args"], dict(details["kwargs"]))) + + @backon.on_predicate( + backon.constant, + lambda v: v != "ok", + max_tries=3, + interval=0, + jitter=None, + on_success=on_success, + ) + def fetch(key, *, source="db"): + return "ok" + + assert fetch("user", source="cache") == "ok" + assert seen == [(("user",), {"source": "cache"})] + + def test_on_predicate_backoff_receives_args(self): + seen = [] + calls = [] + + def on_backoff(details): + seen.append((details["args"], dict(details["kwargs"]))) + + @backon.on_predicate( + backon.constant, + lambda v: v != "ok", + max_tries=3, + interval=0, + jitter=None, + on_backoff=on_backoff, + ) + def fetch(key): + calls.append(key) + return "ok" if len(calls) > 1 else "retry" + + assert fetch("user") == "ok" + assert seen == [(("user",), {})] + + def test_before_sleep_receives_args(self): + seen = [] + + def before_sleep(details): + seen.append((details["args"], dict(details["kwargs"]))) + + @backon.on_exception( + backon.constant, + ValueError, + max_tries=3, + interval=0, + jitter=None, + before_sleep=before_sleep, + ) + def fn(a, *, b=0): + raise ValueError("fail") + + with contextlib.suppress(ValueError): + fn(1, b=2) + + assert len(seen) == 2 + assert all(s == ((1,), {"b": 2}) for s in seen) + + def test_before_and_after_receives_args(self): + seen = [] + + def before(details): + seen.append(("before", details["args"], dict(details["kwargs"]))) + + def after(details): + seen.append(("after", details["args"], dict(details["kwargs"]))) + + @backon.on_exception( + backon.constant, + ValueError, + max_tries=2, + interval=0, + jitter=None, + before=before, + after=after, + ) + def fn(a): + raise ValueError("fail") + + with contextlib.suppress(ValueError): + fn("x") + + assert ("before", ("x",), {}) in seen + assert ("after", ("x",), {}) in seen + + @pytest.mark.asyncio + async def test_async_on_exception_backoff_receives_kwargs(self): + seen = [] + + def handler(details): + seen.append((details["args"], dict(details["kwargs"]))) + + @backon.on_exception( + backon.constant, + ValueError, + max_tries=3, + interval=0, + jitter=None, + on_backoff=handler, + ) + async def do_something(query_id): + raise ValueError("boom") + + with contextlib.suppress(ValueError): + await do_something(query_id=42) + + assert len(seen) == 2 + assert all(args == () and kw == {"query_id": 42} for args, kw in seen) + + @pytest.mark.asyncio + async def test_async_on_predicate_success_receives_args(self): + seen = [] + + def on_success(details): + seen.append((details["args"], dict(details["kwargs"]))) + + @backon.on_predicate( + backon.constant, + lambda v: v != "ok", + max_tries=3, + interval=0, + jitter=None, + on_success=on_success, + ) + async def fetch(key, *, source="db"): + return "ok" + + assert await fetch("user", source="cache") == "ok" + assert seen == [(("user",), {"source": "cache"})] + + def test_retrying_call_state_populates_args_kwargs(self): + r = backon.Retrying( + backon.constant, + exception=ValueError, + max_tries=2, + interval=0, + jitter=None, + ) + + def flaky(x, *, y=0): + raise ValueError("fail") + + with contextlib.suppress(ValueError): + r.call(flaky, 1, y=2) + + cs = r.call_state + assert cs is not None + assert cs.args == (1,) + assert cs.kwargs == {"y": 2} + + def test_retrying_caller_handler_receives_args(self): + seen = [] + + def on_backoff(details): + seen.append((details["args"], dict(details["kwargs"]))) + + caller = backon.RetryingCaller( + backon.constant, + max_tries=3, + jitter=None, + interval=0, + on_backoff=on_backoff, + ) + caller = caller.on(ValueError) + + def flaky(x, *, y=0): + raise ValueError("fail") + + with contextlib.suppress(ValueError): + caller(flaky, 1, y=2) + + assert len(seen) == 2 + assert all(s == ((1,), {"y": 2}) for s in seen) + + @pytest.mark.asyncio + async def test_async_retrying_call_state_populates_args(self): + r = backon.Retrying( + backon.constant, + exception=ValueError, + max_tries=2, + interval=0, + jitter=None, + ) + + async def flaky(x, *, y=0): + raise ValueError("fail") + + with contextlib.suppress(ValueError): + at = r.async_call(flaky, 1, y=2) + await at + + cs = r.call_state + assert cs is not None + assert cs.args == (1,) + assert cs.kwargs == {"y": 2} + + @pytest.mark.asyncio + async def test_async_retrying_caller_handler_receives_args(self): + seen = [] + + def on_backoff(details): + seen.append((details["args"], dict(details["kwargs"]))) + + caller = backon.AsyncRetryingCaller( + backon.constant, + max_tries=3, + jitter=None, + interval=0, + on_backoff=on_backoff, + ) + caller = caller.on(ValueError) + + async def flaky(x, *, y=0): + raise ValueError("fail") + + with contextlib.suppress(ValueError): + await caller(flaky, 1, y=2) + + assert len(seen) == 2 + assert all(s == ((1,), {"y": 2}) for s in seen) + + def test_on_exception_handler_keyerror_regression(self): + captured = {} + + def handler(details): + captured["query_id"] = details["kwargs"]["query_id"] + + @backon.on_exception( + backon.constant, + ValueError, + max_tries=2, + interval=0, + jitter=None, + on_backoff=handler, + ) + def do_something(query_id): + raise ValueError("boom") + + with contextlib.suppress(ValueError): + do_something(query_id=42) + + assert captured == {"query_id": 42} + + +class TestIssue73GeneratorHandlerDetailsArgs: + def test_sync_generator_handler_receives_args(self): + seen = [] + + def on_backoff(details): + seen.append((details["args"], dict(details["kwargs"]))) + + @backon.on_exception( + backon.constant, + ValueError, + max_tries=3, + interval=0, + jitter=None, + on_backoff=on_backoff, + ) + def gen(x, *, y=0): + yield 1 + raise ValueError("fail") + + with contextlib.suppress(ValueError): + list(gen(5, y=9)) + + assert len(seen) == 2 + assert all(s == ((5,), {"y": 9}) for s in seen) + + @pytest.mark.asyncio + async def test_async_generator_handler_receives_args(self): + seen = [] + + def on_backoff(details): + seen.append((details["args"], dict(details["kwargs"]))) + + @backon.on_exception( + backon.constant, + ValueError, + max_tries=3, + interval=0, + jitter=None, + on_backoff=on_backoff, + ) + async def gen(x, *, y=0): + yield 1 + raise ValueError("fail") + + with contextlib.suppress(ValueError): + results = [] + async for item in gen(5, y=9): + results.append(item) + + assert len(seen) == 2 + assert all(s == ((5,), {"y": 9}) for s in seen)