Per-key ordering with partitioned webhook queues

A single ordered queue is the easy answer to message ordering guarantees and the wrong one at scale: it caps your entire delivery throughput at what one consumer can do, and one slow endpoint stalls every customer you have. What you almost always actually need is narrower — every event for one subscription delivered in the order it was produced, with unrelated subscriptions free to run in parallel. This page builds that with partitioned queues in Python, and it deliberately stops short of the global sequencing and reorder buffers described in implementing strict ordering for financial webhooks, which you need only when the ordering contract spans keys.

The mechanism is three rules working together: pick a key, map that key deterministically to one partition, and allow only one consumer on a partition at a time. Break any one of them and ordering evaporates — most commonly the third, when a well-meaning autoscaler adds a second worker to a busy partition. Partitioning also does not eliminate duplicates: the transport is still at-least-once, so the consumer still has to be idempotent, as covered in at-least-once vs exactly-once delivery trade-offs.

Prerequisites

• Python 3.11+ with psycopg 3; the lease implementation uses PostgreSQL session-level advisory locks, and the same shape works on Redis or a broker’s own consumer-group semantics.
• A queue with per-partition FIFO ordering — Kafka topic partitions, an SQS FIFO queue with message group IDs, or a Postgres-backed queue table ordered by an autoincrementing sequence.
• An idempotent consumer, because a redelivery after a lease loss will replay events; see idempotency in webhooks.
• A dead-letter queue reachable from the worker, since ejecting a poison event is the only way to unblock a partition.
• Per-partition lag metrics before you start: without them, head-of-line blocking is invisible until a customer complains.

Step 1: Choose the partition key that matches the ordering contract

The partition key is the ordering contract written as an expression. Everything sharing a key is ordered; everything else is not, and no amount of tuning will give you ordering you did not encode in the key. So the question is not “what identifier is handy” but “what is the narrowest scope across which a consumer will ever compare two events?”

For webhook delivery the answer is usually the subscription — one endpoint belonging to one tenant — because that is the unit whose events a receiver processes in sequence. Choosing the tenant instead is coarser and gives you less parallelism for free; choosing the individual resource is finer and faster, but it breaks the moment a consumer needs order.updated to arrive after order.created for a different resource in the same workflow.

Two properties matter as much as the scope. The key must be stable — present and identical on every event for that entity, including retries and backfills — and it must be high cardinality relative to your partition count, or a handful of large tenants will pin whole partitions while the rest idle.

# partitioning.py
from dataclasses import dataclass
from typing import Any


@dataclass(frozen=True)
class Event:
    event_id: str
    tenant_id: str
    subscription_id: str
    resource_id: str
    payload: dict[str, Any]


def partition_key(event: Event) -> str:
    """The ordering contract, expressed as a key.

    All events for one subscription are ordered relative to each other; nothing
    is ordered across subscriptions. Changing this line changes the guarantee.
    """
    return event.subscription_id

Keeping this in one named function is not ceremony. When someone later asks “are these two events ordered?”, the answer is “only if partition_key returns the same string for both”, and that is a question a reviewer can settle in a diff.

Step 2: Map keys to partitions with a stable hash

Given a key, the partition is hash(key) % partition_count. The hash must be deterministic across processes and restarts, which rules out Python’s built-in hash() for strings — it is randomised per interpreter via PYTHONHASHSEED, so two producers on two pods will disagree about where the same key goes, and the same key ends up in two partitions at once. Use blake2b, md5, or any fixed digest; cryptographic strength is irrelevant, reproducibility is everything.

# routing.py
import hashlib

from partitioning import Event, partition_key

PARTITION_COUNT = 16


def partition_for(key: str, count: int = PARTITION_COUNT) -> int:
    """Deterministic across processes, restarts and language runtimes.

    Do NOT use the builtin hash(): PYTHONHASHSEED randomises it per process,
    so two producers would route the same key to two different partitions.
    """
    digest = hashlib.blake2b(key.encode("utf-8"), digest_size=8).digest()
    return int.from_bytes(digest, "big") % count


def enqueue(cur, event: Event) -> int:
    key = partition_key(event)
    partition = partition_for(key)
    cur.execute(
        "INSERT INTO delivery_queue (partition, partition_key, event_id, payload)"
        " VALUES (%s, %s, %s, %s)",
        (partition, key, event.event_id, event.payload),
    )
    return partition

The queue table itself supplies the ordering inside a partition:

CREATE TABLE delivery_queue (
  seq            BIGSERIAL PRIMARY KEY,   -- FIFO order within a partition
  partition      SMALLINT    NOT NULL,
  partition_key  TEXT        NOT NULL,
  event_id       TEXT        NOT NULL UNIQUE,
  payload        JSONB       NOT NULL,
  attempts       SMALLINT    NOT NULL DEFAULT 0,
  enqueued_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX delivery_queue_drain_idx ON delivery_queue (partition, seq);

Pick the partition count deliberately and expect it to be permanent-ish: it is the modulus, so changing it moves keys, which is why Step 5 exists. A count of 16 to 64 covers most webhook platforms — enough parallelism to saturate a worker fleet, few enough that a per-partition lease and lag metric stay comprehensible.

Step 3: Run exactly one active consumer per partition

Ordering survives only if a partition has one reader. Two workers pulling seq 101 and 102 concurrently will finish in whatever order their HTTP calls complete, and the ordering guarantee you have been paying for is gone — silently, with no error anywhere.

Enforce it with an exclusive lease rather than a convention. A PostgreSQL session-level advisory lock is a good fit: it is held by the connection, so a crashed worker’s lock is released the moment its connection dies, with no lease-expiry timer to tune.

# worker.py
import logging
import time

import psycopg

from failure import handle_failure
from routing import PARTITION_COUNT

logger = logging.getLogger(__name__)

LOCK_NAMESPACE = 4711        # any constant; keeps our locks distinct from other users

CLAIM_HEAD_SQL = """
SELECT seq, event_id, partition_key, payload, attempts
FROM delivery_queue
WHERE partition = %s
ORDER BY seq
LIMIT 1
"""


def try_own_partition(conn: psycopg.Connection, partition: int) -> bool:
    """Session-level advisory lock: exclusive, and auto-released if we die."""
    with conn.cursor() as cur:
        cur.execute("SELECT pg_try_advisory_lock(%s, %s)", (LOCK_NAMESPACE, partition))
        return bool(cur.fetchone()[0])


def release_partition(conn: psycopg.Connection, partition: int) -> None:
    with conn.cursor() as cur:
        cur.execute("SELECT pg_advisory_unlock(%s, %s)", (LOCK_NAMESPACE, partition))


def drain_partition(conn: psycopg.Connection, partition: int, deliver) -> None:
    """Process strictly one event at a time, in seq order. Never parallelise here."""
    while True:
        with conn.cursor() as cur:
            cur.execute(CLAIM_HEAD_SQL, (partition,))
            row = cur.fetchone()
        if row is None:
            return
        seq, event_id, key, payload, attempts = row
        try:
            deliver(key, payload)
        except Exception:
            logger.exception("delivery failed for %s on partition %s", event_id, partition)
            handle_failure(conn, seq, event_id, attempts)
            return          # stop the partition; the head must clear before seq+1
        with conn.cursor() as cur:
            cur.execute("DELETE FROM delivery_queue WHERE seq = %s", (seq,))


def run(conn: psycopg.Connection, deliver) -> None:
    """Each worker sweeps for an unowned partition and holds it until it exits."""
    while True:
        owned = [p for p in range(PARTITION_COUNT) if try_own_partition(conn, p)]
        if not owned:
            time.sleep(1.0)
            continue
        try:
            for partition in owned:
                drain_partition(conn, partition, deliver)
        finally:
            for partition in owned:
                release_partition(conn, partition)
        time.sleep(0.5)

Note what drain_partition does not do: no thread pool, no asyncio.gather, no prefetch of the next event while the current one is in flight. Concurrency inside a partition is the single most common way teams accidentally delete their own ordering guarantee, usually while optimising throughput.

Step 4: Budget for head-of-line blocking

The cost of per-key ordering is head-of-line blocking, and it is not an edge case — it is the design working as specified. If the event at the head of partition 3 cannot be delivered, nothing behind it in partition 3 moves, however healthy those keys are. Partitions 0, 1 and 2 are unaffected, which is exactly the isolation you bought, but every key that hashed to partition 3 now shares one customer’s outage.

Two levers control the damage. More partitions shrink the blast radius, because fewer keys share a partition. And a hard cap on retries at the head, after which the event is ejected to a dead-letter queue, bounds how long a single poison event can hold the line — pair it with exponential backoff with jitter so the attempts themselves do not hammer a recovering endpoint.

# failure.py
import logging

import psycopg

logger = logging.getLogger(__name__)

MAX_ATTEMPTS = 5

DLQ_SQL = """
WITH moved AS (
    DELETE FROM delivery_queue WHERE seq = %(seq)s
    RETURNING partition, partition_key, event_id, payload, attempts
)
INSERT INTO delivery_dlq (partition, partition_key, event_id, payload, attempts, reason)
SELECT partition, partition_key, event_id, payload, attempts, %(reason)s FROM moved
"""


def handle_failure(conn: psycopg.Connection, seq: int, event_id: str, attempts: int) -> None:
    """Retry at the head a bounded number of times, then eject to unblock the partition."""
    with conn.transaction():
        with conn.cursor() as cur:
            if attempts + 1 < MAX_ATTEMPTS:
                cur.execute(
                    "UPDATE delivery_queue SET attempts = attempts + 1 WHERE seq = %s",
                    (seq,),
                )
                return
            cur.execute(DLQ_SQL, {"seq": seq, "reason": f"exhausted after {attempts + 1} attempts"})
            logger.error("ejected %s to DLQ; partition unblocked", event_id)

Ejecting an event is a deliberate ordering violation: the consumer will now see the events that followed it without ever seeing it. That is the right trade for most webhook platforms — one lost event, surfaced loudly in the dead-letter queue, beats an indefinitely stalled partition — but it must be a conscious decision, alarmed on, and reconciled. If your domain cannot tolerate it, the alternative is to stall and page a human, and the timeline for that decision is one you should agree before the incident, not during it.

Step 5: Repartition without breaking ordering

Changing PARTITION_COUNT changes the modulus, so keys move. If producers start writing key sub_42 to partition 9 while its older events are still queued on partition 3, two consumers process that key’s events concurrently and the ordering guarantee is broken for exactly as long as the backlog lasts — the worst kind of outage, because it produces no errors.

The safe procedure is drain-and-switch, and it requires a brief write pause:

1. Freeze the mapping. Deploy the new partition count as configuration only, still inactive, so every producer holds the old modulus.
2. Stop producers from enqueueing, or buffer new events upstream in the outbox.
3. Drain to empty. Wait until every partition reports zero depth; the consumers are still running with the old mapping and finish the backlog in order.
4. Flip the modulus and restart producers and consumers together.
5. Verify that no key appears under two partitions in the queue table.

# repartition.py
import time

import psycopg


def queue_depth(conn: psycopg.Connection) -> int:
    with conn.cursor() as cur:
        cur.execute("SELECT count(*) FROM delivery_queue")
        return int(cur.fetchone()[0])


def wait_until_drained(conn: psycopg.Connection, timeout_s: float = 900.0) -> bool:
    """Block until every partition is empty; a non-empty queue means DO NOT switch."""
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        depth = queue_depth(conn)
        if depth == 0:
            return True
        print(f"waiting to drain: {depth} event(s) still queued")
        time.sleep(5.0)
    return False


def assert_no_split_keys(conn: psycopg.Connection) -> None:
    """After the switch, a key living on two partitions means ordering is broken."""
    with conn.cursor() as cur:
        cur.execute(
            "SELECT partition_key, count(DISTINCT partition) AS n"
            " FROM delivery_queue GROUP BY partition_key HAVING count(DISTINCT partition) > 1"
        )
        offenders = cur.fetchall()
    if offenders:
        raise RuntimeError(f"{len(offenders)} key(s) split across partitions: {offenders[:5]}")

If a write pause is unacceptable, the alternative is to over-provision partitions once and never change the modulus — assign partitions to workers rather than resizing the partition space. Sixty-four partitions with four workers each owning sixteen scales to sixty-four workers without ever rehashing a key, and that is why most platforms pick a large-enough count on day one.

Verification and testing

The test that proves the guarantee is an ordering assertion under concurrency: enqueue interleaved events for several keys, run the workers, and assert that each key’s events were delivered in enqueue order.

# test_ordering.py
import threading
from collections import defaultdict

from routing import partition_for
from worker import drain_partition, try_own_partition


def test_same_key_always_maps_to_one_partition():
    key = "sub_42"
    assert len({partition_for(key) for _ in range(1000)}) == 1


def test_per_key_order_is_preserved_under_concurrency(conn_factory, enqueue_events):
    keys = [f"sub_{i}" for i in range(20)]
    expected = enqueue_events(keys, per_key=25)     # returns {key: [event_id, ...]}

    seen: dict[str, list[str]] = defaultdict(list)
    lock = threading.Lock()

    def deliver(key: str, payload: dict) -> None:
        with lock:
            seen[key].append(payload["event_id"])

    def worker() -> None:
        conn = conn_factory()
        for partition in range(16):
            if try_own_partition(conn, partition):
                drain_partition(conn, partition, deliver)

    threads = [threading.Thread(target=worker) for _ in range(8)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    for key in keys:
        assert seen[key] == expected[key], f"order broken for {key}"

In production the equivalent assertion is a metric, not a test. Export per-partition depth and the age of the oldest queued event, then alert on the age rather than the depth — a deep partition draining quickly is healthy, while a shallow partition whose head has been stuck for ten minutes is an outage for every key on it:

SELECT partition,
       count(*)                                  AS depth,
       max(now() - enqueued_at)                  AS oldest_wait,
       count(DISTINCT partition_key)             AS keys_affected
FROM delivery_queue
GROUP BY partition
ORDER BY oldest_wait DESC;

Failure modes and gotchas

• Using a randomised hash for the mapping. Python’s builtin hash() on a string is seeded per process, so producer pods disagree about the partition for the same key, and a single key ends up spread across partitions with no ordering at all. It looks fine on one machine and breaks the moment you scale out. Always use a fixed digest such as blake2b.
• An autoscaler adding a second consumer to a hot partition. Scaling on queue depth is intuitive and, here, actively harmful: a second reader on one partition destroys the guarantee it exists to provide. Scale the number of partitions owned per worker, never the number of workers per partition, and let the advisory lock reject the extra reader rather than trusting configuration.
• Skew from a whale tenant. One customer generating half your events will pin a partition regardless of how many you have, because their key is atomic. Detect it with count(DISTINCT partition_key) per partition and event counts per key; the fix is a finer key for that tenant only, or a dedicated partition assigned by an override table rather than the hash.
• Retrying the head forever. Without MAX_ATTEMPTS, a permanently failing event blocks its partition indefinitely, and because unrelated partitions stay healthy, dashboards look fine. Cap attempts, eject to the dead-letter queue, and alert on oldest-event age per partition so the stall is visible before a customer reports it.

Frequently Asked Questions

Related

• Implementing strict ordering for financial webhooks — global sequencing and reorder buffers, for when ordering must span keys.
• At-least-once vs exactly-once delivery trade-offs — why partitioned ordering still needs an idempotent consumer.
• Idempotency in webhooks — suppressing the duplicates a lease handover produces.
• Message ordering guarantees — the ordering models this partitioning scheme implements.