acemq_amqp.patterns.ordered

  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"""Keeping one entity's messages in order while everything else runs at once.
 16
 17A queue delivers in order and a consumer with ``concurrency`` above one stops
 18honouring that. Usually the right trade; wrong where a later message about the
 19same thing must not overtake an earlier one — an "order cancelled" handled
 20before the "order placed" it cancels leaves an order that exists and should not.
 21
 22The answer is not to give up concurrency, because the messages that must not
 23overtake each other are only the ones about the same entity. Ordering per key
 24and concurrency across keys is what this buys.
 25"""
 26
 27from __future__ import annotations
 28
 29import asyncio
 30from collections.abc import AsyncIterator, Callable
 31from contextlib import asynccontextmanager
 32from typing import TypeAlias
 33
 34from ..ack import Ack
 35from ..connection import AsyncHandler, Handler, Message
 36from ._support import decide
 37
 38#: The FNV-1a 32-bit offset basis and prime. Named rather than inlined because
 39#: the two constants are the whole of the agreement between languages.
 40_FNV_OFFSET = 2166136261
 41_FNV_PRIME = 16777619
 42_FNV_MASK = 0xFFFFFFFF
 43
 44#: Picks the ordering key for a message. Messages sharing a key are handled one
 45#: at a time and in the order they arrived; messages with different keys may be
 46#: handled at once.
 47PartitionKey: TypeAlias = Callable[[Message], str]
 48
 49
 50def by_header(name: str) -> PartitionKey:
 51    """Orders by an application header — a tenant, a customer, an aggregate.
 52
 53    :param name: which header carries the entity's identity
 54    :returns: a key function
 55    """
 56
 57    def key(message: Message) -> str:
 58        value = message.envelope.headers.get(name)
 59        return "" if value is None else str(value)
 60
 61    return key
 62
 63
 64def by_correlation() -> PartitionKey:
 65    """Orders by correlation identifier, which keeps one business action's
 66    messages in sequence."""
 67
 68    def key(message: Message) -> str:
 69        return message.envelope.correlation_id
 70
 71    return key
 72
 73
 74def ordered(key: PartitionKey, handler: Handler) -> AsyncHandler:
 75    """Wraps a handler so messages sharing a key are never handled at once::
 76
 77        consumer = await mq.consume(
 78            "orders", ordered(by_header("order-id"), handle), concurrency=16
 79        )
 80
 81    **What it does not do.** It orders the handling of messages that have
 82    already been delivered. It cannot reorder ones the broker delivered out of
 83    order, and with several consumers on one queue it orders only within each
 84    process — ordering across processes needs the messages to reach the same
 85    process in the first place, which is a routing decision rather than a
 86    handler one. :func:`partitioned_routing_key` is for making that decision.
 87
 88    A message whose key is empty is handled with no ordering at all, because
 89    there is nothing to order it against. That is a deliberate choice over
 90    treating "no key" as one shared key, which would serialise every message
 91    that happens to be missing the header — turning a producer's bug into a
 92    consumer's outage.
 93
 94    :param key: what identifies the entity a message is about
 95    :param handler: what to do with the message
 96    :returns: the wrapped handler
 97    """
 98    locks = _KeyedLocks()
 99
100    async def in_order(message: Message) -> Ack:
101        identity = key(message)
102        if not identity:
103            return await decide(handler, message)
104        async with locks.hold(identity):
105            return await decide(handler, message)
106
107    return in_order
108
109
110def partition(key: str, count: int) -> int:
111    """Maps a key onto one of ``count`` slots, the same way in every language.
112
113    For deciding which queue or which worker a message belongs to, when the
114    ordering has to hold across processes rather than within one. FNV-1a rather
115    than :func:`hash`, because Python's string hash is randomised per process by
116    default: two workers in the same deployment would disagree about where a key
117    belongs, which is the one thing a partition function must not do.
118
119    :param key: what identifies the entity
120    :param count: how many slots there are
121    :returns: the slot, from 0
122    """
123    if count <= 1:
124        return 0
125    digest = _FNV_OFFSET
126    for byte in key.encode("utf-8"):
127        digest ^= byte
128        digest = (digest * _FNV_PRIME) & _FNV_MASK
129    return digest % count
130
131
132def partitioned_routing_key(base: str, key: str, partitions: int) -> str:
133    """``orders`` and an order id become ``orders.3``.
134
135    For publishing into a queue-per-partition arrangement, which is how ordering
136    is made to hold across processes: every message about one entity is routed
137    to one queue, and one queue is consumed in order.
138
139    :param base: the routing key without the partition
140    :param key: what identifies the entity
141    :param partitions: how many there are
142    :returns: the routing key
143    """
144    return f"{base}.{partition(key, partitions)}"
145
146
147class _KeyedLocks:
148    """One lock per key, forgotten once nothing is holding it.
149
150    The forgetting is the point. A key is usually an order or a customer, so a
151    map that kept one lock per key ever seen would grow for the life of the
152    process — the leak that only shows up in the service that has been running
153    longest.
154    """
155
156    def __init__(self) -> None:
157        self._locks: dict[str, asyncio.Lock] = {}
158        self._holders: dict[str, int] = {}
159
160    @asynccontextmanager
161    async def hold(self, key: str) -> AsyncIterator[None]:
162        lock = self._locks.get(key)
163        if lock is None:
164            lock = self._locks[key] = asyncio.Lock()
165        # Counted before the wait rather than after it, so a key with somebody
166        # queued on it is not forgotten by the holder that releases first.
167        self._holders[key] = self._holders.get(key, 0) + 1
168        try:
169            await lock.acquire()
170        except BaseException:
171            self._let_go(key)
172            raise
173        try:
174            yield
175        finally:
176            lock.release()
177            self._let_go(key)
178
179    def _let_go(self, key: str) -> None:
180        remaining = self._holders[key] - 1
181        if remaining:
182            self._holders[key] = remaining
183            return
184        del self._holders[key]
185        del self._locks[key]
186
187    def __len__(self) -> int:
188        """How many keys are being held, for a test that wants to know the
189        forgetting works."""
190        return len(self._locks)