diff --git a/RELEASE.md b/RELEASE.md index b9674717..5265fdc2 100644 --- a/RELEASE.md +++ b/RELEASE.md @@ -96,6 +96,12 @@ failed, was held or was retried), and the web dashboard's failures and job pages returned 500 when its router was mounted on psycopg queries. `Job.headers` already accepted both forms. +- `DatabaseRetryEntrypointExecutor` caps a backoff too large for a + `timedelta` at `max_delay`. The delay was multiplied out before the cap, so + once `initial_delay * backoff_multiplier**attempts` passed timedelta's range + (attempt 47 with the defaults, attempt 14 with a multiplier of 10) the + executor raised `OverflowError`, and the job ended as `exception` (or + `failed` under `on_failure="hold"`) with attempts left in `max_attempts`. ## v1.6.0 diff --git a/pgqueuer/core/executors.py b/pgqueuer/core/executors.py index 00210d97..5a9e0436 100644 --- a/pgqueuer/core/executors.py +++ b/pgqueuer/core/executors.py @@ -120,10 +120,14 @@ async def execute(self, job: models.Job, context: models.Context) -> None: raise retry from e if job.attempts >= self.max_attempts: raise - delay = min( - self.initial_delay * (self.backoff_multiplier**job.attempts), - self.max_delay, - ) + try: + delay = min( + self.initial_delay * (self.backoff_multiplier**job.attempts), + self.max_delay, + ) + except OverflowError: + # The uncapped delay no longer fits a timedelta (or a float), so the cap applies. + delay = self.max_delay raise errors.RetryRequested(delay=delay, reason=str(e)) from e diff --git a/test/test_retry_requeue.py b/test/test_retry_requeue.py index c34458ad..b24386b6 100644 --- a/test/test_retry_requeue.py +++ b/test/test_retry_requeue.py @@ -549,6 +549,60 @@ def test_database_retry_executor_backoff_caps_at_max_delay() -> None: assert len(set(delays)) > 1 +@pytest.mark.parametrize( + ("backoff_multiplier", "attempts", "expected"), + [ + pytest.param(2.0, 3, timedelta(seconds=8), id="x2-below-cap"), + pytest.param(2.0, 46, timedelta(minutes=5), id="x2-last-in-timedelta-range"), + pytest.param(2.0, 47, timedelta(minutes=5), id="x2-past-timedelta-range"), + pytest.param(10.0, 14, timedelta(minutes=5), id="x10-past-timedelta-range"), + pytest.param(2.0, 1100, timedelta(minutes=5), id="x2-past-float-range"), + ], +) +async def test_database_retry_executor_caps_delay_past_timedelta_range( + backoff_multiplier: float, + attempts: int, + expected: timedelta, +) -> None: + """A backoff too large for a timedelta is capped at max_delay instead of raising.""" + + async def handler(job: Job) -> None: + raise ValueError("still failing") + + executor = DatabaseRetryEntrypointExecutor( + parameters=EntrypointExecutorParameters( + concurrency_limit=0, + func=handler, + ), + max_attempts=attempts + 1, + initial_delay=timedelta(seconds=1), + max_delay=timedelta(minutes=5), + backoff_multiplier=backoff_multiplier, + ) + job = Job.model_validate( + { + "id": JobId(1), + "priority": 0, + "created": "2024-01-01T00:00:00Z", + "updated": "2024-01-01T00:00:00Z", + "heartbeat": "2024-01-01T00:00:00Z", + "execute_after": "2024-01-01T00:00:00Z", + "status": "picked", + "entrypoint": QueueEntrypoint("overflow_ep"), + "payload": None, + "attempts": attempts, + "queue_manager_id": None, + "headers": None, + } + ) + + with pytest.raises(RetryRequested) as excinfo: + await executor.execute(job, Context(cancellation=anyio.CancelScope())) + + assert excinfo.value.delay == expected + assert isinstance(excinfo.value.__cause__, ValueError) + + # --------------------------------------------------------------------------- # Postgres integration # --------------------------------------------------------------------------- @@ -621,6 +675,53 @@ async def handler(job: Job) -> None: assert payloads_seen == [b"critical-data", b"critical-data"] +@pytest.mark.parametrize( + "max_attempts", + [ + pytest.param(10, id="budget-below-overflow"), + pytest.param(20, id="budget-past-overflow"), + ], +) +async def test_database_retry_executor_uses_full_budget_postgres( + apgdriver: AsyncpgDriver, + max_attempts: int, +) -> None: + """Every attempt in max_attempts runs, even once the uncapped backoff would overflow.""" + q = Queries(apgdriver) + qm = QueueManager(Queries(apgdriver)) + seen_attempts: list[int] = [] + + # 1 ms * 10**17 is past timedelta's range, so attempts 17 and up rely on the cap. + @qm.entrypoint( + "pg_long_budget_ep", + executor_factory=lambda params: DatabaseRetryEntrypointExecutor( + parameters=params, + max_attempts=max_attempts, + initial_delay=timedelta(milliseconds=1), + max_delay=timedelta(milliseconds=5), + backoff_multiplier=10.0, + ), + ) + async def handler(job: Job) -> None: + seen_attempts.append(job.attempts) + raise ValueError("still failing") + + (job_id,) = await q.enqueue("pg_long_budget_ep", None, priority=0) + + async with asyncio.timeout(15): + await qm.run( + mode=QueueExecutionMode.drain, + dequeue_timeout=timedelta(seconds=1), + ) + + assert seen_attempts == list(range(max_attempts + 1)) + (terminal,) = [ + log for log in await q.queue_log() if log.job_id == job_id and log.status == "exception" + ] + assert terminal.traceback is not None + assert terminal.traceback.exception_type == "ValueError" + + # --------------------------------------------------------------------------- # Edge cases — InMemory # ---------------------------------------------------------------------------