acemq_amqp.retry

  1# Copyright 2026 AceMQ.
  2#
  3# Licensed under the Apache License, Version 2.0 (the "License");
  4# you may not use this file except in compliance with the License.
  5# You may obtain a copy of the License at
  6#
  7#     https://www.apache.org/licenses/LICENSE-2.0
  8#
  9# Unless required by applicable law or agreed to in writing, software
 10# distributed under the License is distributed on an "AS IS" BASIS,
 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12# See the License for the specific language governing permissions and
 13# limitations under the License.
 14
 15"""When to try again, where to wait, and when to stop.
 16
 17The schedule arithmetic here is part of the cross-language contract rather than
 18a local choice: the same policy must produce the same delays in Java, Go, .NET
 19and Python, because the same message can be retried by a consumer written in
 20any of them. The jitter is the one exception — it is random by definition —
 21so :meth:`RetryPolicy.schedule` exposes the delays *without* it, which is what
 22to read when deciding whether a policy is the one you meant.
 23
 24Where the waiting happens is the other half. A short wait is spent in the
 25consumer, holding the delivery; a long one is spent in the broker, in a queue
 26whose message TTL is the wait. The threshold between them is
 27:data:`DEFAULT_BROKER_WAIT_THRESHOLD` unless a policy says otherwise, and the
 28reason for having a threshold at all rather than one rule is in
 29:class:`RetryPolicy`.
 30"""
 31
 32from __future__ import annotations
 33
 34import random
 35from dataclasses import dataclass, replace
 36from datetime import timedelta
 37
 38ZERO = timedelta(0)
 39
 40#: Waits this long or longer are spent in the broker rather than in the
 41#: consumer. Thirty seconds is roughly where the two costs cross: below it, the
 42#: seconds a restart loses are only seconds and a held prefetch slot is cheap;
 43#: above it, a consumer that restarts mid-wait loses the wait entirely, because
 44#: the broker redelivers the unacknowledged message at once and a five-minute
 45#: backoff becomes instant.
 46DEFAULT_BROKER_WAIT_THRESHOLD = timedelta(seconds=30)
 47
 48
 49@dataclass(frozen=True, slots=True)
 50class Wait:
 51    """How long before the next attempt, and where the message spends it.
 52
 53    Two answers rather than one because they cannot be worked out separately:
 54    jitter applies only to a wait spent in the consumer, so a caller given a
 55    delay alone could not tell whether it had already been moved — and a
 56    jittered delay does not name a rung queue.
 57
 58    :param delay: how long the message waits
 59    :param in_broker: whether it waits on a rung queue rather than here
 60    """
 61
 62    delay: timedelta
 63    in_broker: bool
 64
 65
 66@dataclass(frozen=True, slots=True)
 67class RetryPolicy:
 68    """How many times to try, how long to wait, and where.
 69
 70    Waits below ``broker_wait_threshold`` are spent in the consumer, which holds
 71    the delivery and one prefetch slot for the duration. Waits at or above it
 72    are spent in a ``{queue}.retry.{delay}`` queue whose ``x-message-ttl`` is the
 73    wait and whose dead-letter target is the source queue, so the broker returns
 74    the message when the time is up and a consumer restart cannot shorten it.
 75
 76    Splitting at a threshold rather than picking one of the two takes the
 77    durability where it is worth its complexity and leaves the simplicity where
 78    it is not. A queue's TTL is fixed at declaration and so cannot express a
 79    jittered delay, which is why jitter applies below the threshold only; above
 80    it the spread comes free, because each message's TTL starts when it enters
 81    the rung, so a fleet that failed over ten seconds is released over ten
 82    seconds.
 83
 84    :param max_attempts: total deliveries including the first. 1 means no retry
 85    :param initial_delay: the wait before the second attempt
 86    :param multiplier: what the delay is multiplied by each time
 87    :param max_delay: the ceiling, or zero for none
 88    :param jitter_factor: how far a delay may move either side, 0 to 1
 89    :param max_message_age: give up on anything older, or zero for never
 90    :param broker_wait_threshold: waits this long or longer are spent in the
 91        broker, or zero to spend every wait in the consumer
 92    """
 93
 94    max_attempts: int = 1
 95    initial_delay: timedelta = ZERO
 96    multiplier: float = 2.0
 97    max_delay: timedelta = ZERO
 98    jitter_factor: float = 0.0
 99    max_message_age: timedelta = ZERO
100    broker_wait_threshold: timedelta = DEFAULT_BROKER_WAIT_THRESHOLD
101
102    def give_up_after(self, age: timedelta) -> RetryPolicy:
103        """A copy that abandons a message older than ``age``.
104
105        The honest limit when a queue has been paused: attempts say nothing
106        about how long a message has been waiting, and a message four days old
107        is usually one nobody wants delivered now.
108        """
109        return replace(self, max_message_age=age)
110
111    def with_jitter(self, factor: float) -> RetryPolicy:
112        """A copy using a different jitter factor, between 0 and 1."""
113        return replace(self, jitter_factor=factor)
114
115    def wait_in_broker_from(self, threshold: timedelta) -> RetryPolicy:
116        """A copy that moves the line between waiting here and waiting there.
117
118        Zero is the way out: with no threshold nothing is long enough to reach
119        the broker, so every wait is spent in the consumer and no rung queue is
120        needed. That is the right setting for a service whose broker it may not
121        declare queues on, and the wrong one for a policy with delays measured
122        in minutes.
123
124        :param threshold: waits this long or longer go to a rung queue, or zero
125            for none of them
126        :returns: a new policy
127        """
128        return replace(self, broker_wait_threshold=threshold)
129
130    def waits_in_broker(self, delay: timedelta) -> bool:
131        """Whether a wait of this length belongs on a rung queue.
132
133        :param delay: an unjittered delay, as :meth:`schedule` reports them
134        :returns: whether the broker rather than the consumer should hold it
135        """
136        return (
137            self.broker_wait_threshold > ZERO
138            and delay > ZERO
139            and delay >= self.broker_wait_threshold
140        )
141
142    def broker_rungs(self) -> list[timedelta]:
143        """The delays this policy needs a rung queue for, longest last.
144
145        Exactly the entries of :meth:`schedule` that are at or above the
146        threshold, with repeats removed — a fixed policy that waits a minute
147        three times needs one queue, not three. It is a finite list because the
148        schedule is, which is what makes the queues declarable up front rather
149        than conjured by a consumer at the moment it first fails.
150
151        :returns: one delay per queue, in the order the schedule reaches them
152        """
153        return list(dict.fromkeys(d for d in self.schedule() if self.waits_in_broker(d)))
154
155    def next_delay(self, attempt: int, message_age: timedelta = ZERO) -> timedelta | None:
156        """How long to wait before the next attempt, or ``None`` to give up.
157
158        :param attempt: the attempt that has just failed, starting at 1
159        :param message_age: how old the message is
160        :returns: the delay, or ``None`` when there should be no further attempt
161        """
162        wait = self.next_wait(attempt, message_age)
163        return None if wait is None else wait.delay
164
165    def next_wait(self, attempt: int, message_age: timedelta = ZERO) -> Wait | None:
166        """The whole answer: how long to wait, and where.
167
168        :param attempt: the attempt that has just failed, starting at 1
169        :param message_age: how old the message is
170        :returns: the wait, or ``None`` when there should be no further attempt
171        """
172        if attempt >= self.max_attempts:
173            return None
174        if self.max_message_age > ZERO and message_age >= self.max_message_age:
175            return None
176
177        delay = self._unjittered(attempt)
178
179        if self.waits_in_broker(delay):
180            # Deliberately not jittered. A rung queue's TTL is fixed when it is
181            # declared, so a moved delay would name a queue that does not exist;
182            # and the spread jitter buys is already there, because each message's
183            # TTL starts when it arrives rather than when the batch failed.
184            return Wait(delay, in_broker=True)
185
186        if self.jitter_factor > 0 and delay > ZERO:
187            # Both directions, so a fleet of consumers that failed together
188            # does not come back together. One-sided jitter only ever delays,
189            # which turns a thundering herd into a slower thundering herd.
190            factor = 1 + ((random.random() * 2 - 1) * self.jitter_factor)
191            delay = delay * factor
192
193        return Wait(max(delay, ZERO), in_broker=False)
194
195    def _unjittered(self, attempt: int) -> timedelta:
196        """The delay after ``attempt``, before jitter and before the threshold."""
197        delay = self.initial_delay
198        for _ in range(1, attempt):
199            delay = delay * self.multiplier
200            if self.max_delay > ZERO and delay > self.max_delay:
201                delay = self.max_delay
202                break
203        if self.max_delay > ZERO and delay > self.max_delay:
204            delay = self.max_delay
205        return delay
206
207    def schedule(self) -> list[timedelta]:
208        """The delays this policy would use, without jitter.
209
210        What to look at when deciding whether a policy is the one you meant:
211        ``ExponentialRetry(5, 1s, 1m).schedule()`` is ``[1s, 2s, 4s, 8s]``, and
212        four numbers are easier to argue with than three parameters.
213        """
214        delays: list[timedelta] = []
215        delay = self.initial_delay
216        for _ in range(1, max(self.max_attempts, 0)):
217            capped = delay
218            if self.max_delay > ZERO and capped > self.max_delay:
219                capped = self.max_delay
220            delays.append(capped)
221            delay = delay * self.multiplier
222        return delays
223
224
225def no_retry() -> RetryPolicy:
226    """One delivery, no second chance."""
227    return RetryPolicy(max_attempts=1)
228
229
230def exponential_retry(
231    max_attempts: int,
232    initial_delay: timedelta,
233    max_delay: timedelta = ZERO,
234) -> RetryPolicy:
235    """Doubling delays with 20% jitter, which is the sane default.
236
237    :param max_attempts: total deliveries including the first
238    :param initial_delay: the wait before the second attempt
239    :param max_delay: the ceiling, or zero for none
240    """
241    return RetryPolicy(
242        max_attempts=max_attempts,
243        initial_delay=initial_delay,
244        multiplier=2.0,
245        max_delay=max_delay,
246        jitter_factor=0.2,
247    )
248
249
250def fixed_retry(max_attempts: int, delay: timedelta) -> RetryPolicy:
251    """The same wait every time, with no jitter."""
252    return RetryPolicy(max_attempts=max_attempts, initial_delay=delay, multiplier=1.0)