acemq_amqp.ack

  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"""What a handler says about a message, and what the consumer then did.
 16
 17Two different statements, and the difference is the point of
 18:class:`Settlement`. A handler saying :func:`retry` is a request; whether there
 19is an attempt left to grant it is the retry policy's answer, and a message that
 20asked to be retried once too often is dead-lettered. Anything watching a
 21delivery — the tracing adapter is the one that matters — needs the answer rather
 22than the request, which is why the consumer reports it separately.
 23"""
 24
 25from __future__ import annotations
 26
 27from dataclasses import dataclass
 28from datetime import timedelta
 29from enum import Enum
 30
 31#: The consumer acknowledged the message; the handler was happy with it.
 32OUTCOME_ACKED = "acked"
 33
 34#: The message goes round again, after :attr:`Settlement.delay`.
 35OUTCOME_RETRIED = "retried"
 36
 37#: The handler refused it outright. It is dead-lettered, and the word says the
 38#: decision was deliberate rather than the end of a losing streak.
 39OUTCOME_REJECTED = "rejected"
 40
 41#: It ran out of attempts, aged out, or was unprocessable.
 42OUTCOME_DEAD_LETTERED = "dead_lettered"
 43
 44#: It went to ``{queue}.parked`` rather than to the dead letters, because
 45#: nothing could read it. The same word the ``parked`` outcome on
 46#: ``acemq.messages.dead.lettered.total`` uses, and the word the engine already
 47#: used for a body that would not decode.
 48OUTCOME_PARKED = "parked"
 49
 50#: Somebody else holds a live claim on the message and has not finished, so it
 51#: was put back to be looked at again later — without spending an attempt, and
 52#: never dead-lettered for it. See :func:`in_progress`.
 53OUTCOME_IN_PROGRESS = "in_progress"
 54
 55#: How long a message found in progress waits before it is looked at again.
 56DEFAULT_IN_PROGRESS_DELAY = timedelta(seconds=5)
 57
 58
 59class Action(Enum):
 60    """The things that can be done with a delivered message."""
 61
 62    ACCEPT = "accept"
 63    RETRY = "retry"
 64    REJECT = "reject"
 65    PARK = "park"
 66    IN_PROGRESS = "in_progress"
 67
 68
 69@dataclass(frozen=True, slots=True)
 70class Ack:
 71    """A handler's decision: it worked, try it again, never again, or nobody
 72    can read it.
 73
 74    Returned rather than performed, so a handler that forgets to decide is a
 75    handler that returns ``None`` and is caught immediately, rather than one
 76    whose message sits unacknowledged until the connection drops and then comes
 77    back — usually to the same handler, with the same outcome.
 78    """
 79
 80    action: Action
 81    error: BaseException | None = None
 82    #: How long to wait before the message is looked at again. Read for
 83    #: :attr:`Action.IN_PROGRESS` only; a retry's wait is the policy's.
 84    delay: timedelta | None = None
 85
 86    def __str__(self) -> str:
 87        return self.action.value
 88
 89
 90@dataclass(frozen=True, slots=True)
 91class Settlement:
 92    """What the consumer did with a delivery, once it had decided.
 93
 94    Reported before it is carried out rather than after, because one of them
 95    takes time: a consumer-side retry sleeps out the backoff before the message
 96    is republished, and a listener told afterwards would be describing the
 97    delivery minutes later, on a span that had been open the whole while.
 98
 99    :param outcome: one of :data:`OUTCOME_ACKED`, :data:`OUTCOME_RETRIED`,
100        :data:`OUTCOME_REJECTED`, :data:`OUTCOME_DEAD_LETTERED` and
101        :data:`OUTCOME_PARKED` — the same words the other libraries put on a
102        span
103    :param reason: why it was set aside, when it was
104    :param delay: how long until the next attempt, when there is one
105    """
106
107    outcome: str
108    reason: str | None = None
109    delay: timedelta | None = None
110
111    @property
112    def dead_lettered(self) -> bool:
113        """Whether the message went to the dead-letter queue.
114
115        True for an outright rejection as well as for an exhausted one: both
116        end up in the same queue, and the difference between the two words is
117        *who decided*, which :attr:`outcome` records and this does not.
118
119        False for :data:`OUTCOME_PARKED`, and that is the whole point of the
120        parked queue: a message nobody could read is a different problem with a
121        different answer from one that failed five times, and whoever drains
122        the dead letters should not have to sort the two apart by hand.
123        """
124        return self.outcome in (OUTCOME_REJECTED, OUTCOME_DEAD_LETTERED)
125
126    @property
127    def parked(self) -> bool:
128        """Whether the message went to the parked queue."""
129        return self.outcome == OUTCOME_PARKED
130
131    def __str__(self) -> str:
132        return self.outcome
133
134
135def accept() -> Ack:
136    """Confirms the message. It will not be delivered again."""
137    return Ack(Action.ACCEPT)
138
139
140def retry(error: BaseException | None = None) -> Ack:
141    """Returns the message to be tried again.
142
143    The retry policy decides whether there is another attempt left; when there
144    is not, the message is dead-lettered with ``error`` as the reason. A
145    :class:`Fatal` error skips the remaining attempts, because they would all
146    fail the same way.
147    """
148    return Ack(Action.RETRY, error)
149
150
151def reject(error: BaseException | None = None) -> Ack:
152    """Dead-letters the message without trying again.
153
154    For when the message itself is the problem — a field that cannot be missing
155    is missing, a reference points at nothing — rather than when the world is
156    temporarily unhelpful.
157    """
158    return Ack(Action.REJECT, error)
159
160
161def park(error: BaseException | None = None) -> Ack:
162    """Sets the message aside on ``{queue}.parked`` without trying again.
163
164    For when the message is not readable at all: a body in a format nothing
165    here understands, a version this service was never taught. The engine
166    already parks a body its codec cannot decode; this is how a handler that
167    worked out the same thing one layer further in says so.
168
169    :func:`reject` is the neighbouring word and the difference is worth
170    keeping. A rejected message is a *bad request* — understood, and refused —
171    and it belongs with the other dead letters. A parked one was never
172    understood, and mixing the two means whoever drains the dead-letter queue
173    has to sort them by hand to find the producer that is emitting rubbish.
174    """
175    return Ack(Action.PARK, error)
176
177
178def in_progress(delay: timedelta = DEFAULT_IN_PROGRESS_DELAY) -> Ack:
179    """Puts the message back because somebody else is still working on it.
180
181    What :func:`~acemq_amqp.patterns.idempotent` answers when a redelivery finds
182    a claim that is live but not yet confirmed. Accepting it would lose the
183    message if that other handler then died; running it would do the work
184    twice. So it goes back on its own queue after ``delay``, with its attempt
185    **unchanged**: a claim held elsewhere says nothing about whether this
186    message can be handled, so it neither spends a retry nor can exhaust the
187    policy and be dead-lettered. When the claim is confirmed the next look is a
188    duplicate and is accepted; when its lease runs out, it is taken over.
189
190    The wait is spent in the consumer, holding the delivery, as a short retry's
191    is; keep it short beside the store's claim timeout.
192
193    :param delay: how long until the message is looked at again
194    """
195    return Ack(Action.IN_PROGRESS, delay=delay)
196
197
198class FatalError(Exception):
199    """An error no number of retries will fix.
200
201    Wrapping a cause in this says "stop now" without the handler having to know
202    how many attempts remain, which is knowledge handlers should not need.
203
204    The other languages call this ``Fatal``; the name is Python's because a
205    class name is ergonomics rather than wire contract, and Python expects an
206    exception to say that it is one.
207    """