Partitioning Webhook Queues by Tenant Without Shard Explosion

At 09:14 a customer imports two million records and your platform faithfully generates two million webhooks. At 09:16 a different customer’s password-reset webhook, which normally arrives in under a second, is sitting eleven minutes deep behind that import. Nothing is broken — no endpoint is down, no worker is crashing, the error rate is flat — and yet for most of your customers the product is now unusable. This is head-of-line blocking, and it is the failure mode that the topology decisions in webhook delivery queue architecture exist to bound.

The instinct is to give every tenant its own queue. That works beautifully at 50 tenants and collapses at 40,000, so this guide walks the path most platforms end up on instead: measure first, partition by a hashed key, schedule fairly inside each partition, and reserve genuinely dedicated lanes for the handful of accounts that earn them. Broker support for keyed partitioning varies a lot, so if you have not yet committed, read choosing a message broker for webhook delivery alongside this.

Prerequisites

• A tenant_id on every envelope, set at enqueue and never derived later from the payload.
• Redis 7 (or an equivalent) for lane state and counters, with redis-py 5.0 and Python 3.11.
• Per-tenant metrics you can already emit: enqueued count, delivered count, and delivery duration. Duration is not optional — fairness measured in messages is the wrong fairness.
• A queue that supports either keyed partitioning or many logical lanes. The examples use Redis lists per lane; the same design works over SQS with one queue per shard.
• A defined ordering key. If events for the same tenant must stay ordered, resharding becomes a drain-and-move operation, as covered in per-key ordering with partitioned queues.

Step 1: Attribute the backlog to a tenant before changing anything

Brokers report queue depth. They do not report whose depth it is, and without that number every conversation about isolation is speculation. Maintain two counters per tenant at the point of enqueue and completion: outstanding envelopes, and cumulative worker-seconds consumed. The second one is what actually matters. A tenant sending 5,000 events an hour to an endpoint that answers in 40 ms is a rounding error; a tenant sending 500 an hour to an endpoint that takes 25 seconds to time out is consuming 200 times more of your delivery capacity per message.

"""Per-tenant backlog and capacity attribution."""
import time

import redis

R = redis.Redis.from_url("redis://localhost:6379/0")
DEPTH = "wh:tenant:depth"        # HASH tenant -> outstanding envelopes
SECONDS = "wh:tenant:seconds"    # HASH tenant -> cumulative worker-seconds


def on_enqueue(tenant: str) -> None:
    R.hincrby(DEPTH, tenant, 1)


def on_complete(tenant: str, elapsed: float) -> None:
    pipe = R.pipeline()
    pipe.hincrby(DEPTH, tenant, -1)
    pipe.hincrbyfloat(SECONDS, tenant, elapsed)
    pipe.execute()


def top_consumers(limit: int = 10) -> list[tuple[str, int, float, float]]:
    """Rank tenants by share of outstanding work and of consumed capacity."""
    depth = {k.decode(): int(v) for k, v in R.hgetall(DEPTH).items()}
    seconds = {k.decode(): float(v) for k, v in R.hgetall(SECONDS).items()}
    total_depth = sum(depth.values()) or 1
    total_seconds = sum(seconds.values()) or 1.0
    rows = [
        (
            tenant,
            count,
            count / total_depth,
            seconds.get(tenant, 0.0) / total_seconds,
        )
        for tenant, count in depth.items()
    ]
    rows.sort(key=lambda row: row[3], reverse=True)
    return rows[:limit]


if __name__ == "__main__":
    print(f"{'tenant':<20}{'depth':>10}{'depth %':>10}{'capacity %':>12}")
    for tenant, count, depth_share, capacity_share in top_consumers():
        print(f"{tenant:<20}{count:>10,}{depth_share:>9.1%}{capacity_share:>11.1%}")

Run this during an incident and the shape of the problem becomes obvious in one line of output. In most webhook platforms the top tenant holds somewhere between 40% and 80% of outstanding work at peak. The number that justifies engineering effort is not the whale’s share, though — it is the p95 queue wait for everyone else while the whale is active.

Step 2: Partition by hashed tenant key instead of one queue per tenant

Queue-per-tenant is the obvious design and it fails on arithmetic rather than on principle. Every queue needs a consumer, and an idle consumer still costs something: on SQS, 40,000 queues polled at a 20-second long-poll interval generate about 5.2 billion ReceiveMessage calls a month — roughly $2,000 in request charges to discover that nothing happened. On RabbitMQ each queue is a live process with its own memory and its own consumer channel, and tens of thousands of them turn broker restarts into multi-minute events. On Kafka, a partition per tenant blows past the few thousand partitions per broker that rebalance in a reasonable time. That is shard explosion: the cost is not per message, it is per lane, and it is paid whether the lane is busy or empty.

A fixed shard count decouples the two. Pick a number that comfortably exceeds your worker group count — 32 or 64 is typical — and hash the tenant into it. Use rendezvous hashing rather than hash(tenant) % n, because modulo remaps nearly every tenant when the shard count changes, whereas rendezvous moves only the fraction that must move.

"""Route envelopes to a fixed set of shards using rendezvous (highest-random-weight) hashing."""
import hashlib
import json

import redis

R = redis.Redis.from_url("redis://localhost:6379/0")
SHARD_COUNT = 32
PIN = "wh:tenant:pin"        # HASH tenant -> shard, for manual overrides


def _weight(tenant: str, shard: int) -> int:
    digest = hashlib.blake2b(f"{tenant}:{shard}".encode(), digest_size=8).digest()
    return int.from_bytes(digest, "big")


def shard_for(tenant: str, shard_count: int = SHARD_COUNT) -> int:
    """Rendezvous hashing: changing shard_count moves ~1/n of tenants, not all of them."""
    pinned = R.hget(PIN, tenant)
    if pinned is not None:
        return int(pinned)
    return max(range(shard_count), key=lambda shard: _weight(tenant, shard))


def enqueue(envelope: dict) -> int:
    shard = shard_for(envelope["tenant"])
    R.lpush(f"wh:shard:{shard}", json.dumps(envelope))
    return shard


def shard_distribution(tenants: list[str], shard_count: int = SHARD_COUNT) -> dict[int, int]:
    counts = dict.fromkeys(range(shard_count), 0)
    for tenant in tenants:
        counts[shard_for(tenant, shard_count)] += 1
    return counts

The PIN override is what makes Step 4 possible: a tenant can be moved to a specific shard without touching the hash function.

Be honest about what this buys. Hash partitioning divides the blast radius by the shard count; it does not eliminate it. With 32 shards, the whale still saturates one shard completely, and the roughly 1/32 of your tenants who hash into that shard experience exactly the same eleven-minute wait as before. They are simply fewer, and they are randomly selected — which is worse to explain to a customer, not better. Partitioning is the substrate; the fairness work in Step 3 is what actually fixes it.

Step 3: Add weighted fair scheduling inside each shard

Inside a shard, stop treating the queue as FIFO across tenants. Give each active tenant its own lane and serve the lanes in rotation, with a credit balance denominated in worker-seconds so a tenant with slow endpoints cannot quietly consume the shard’s whole capacity while looking well-behaved on a message count.

The three moving parts are a ring of active tenants, a per-tenant lane, and a credit ledger refilled proportionally to each tenant’s weight. A tenant with no credit is skipped until the next refill; a tenant with an empty lane leaves the ring.

"""Weighted fair scheduling across tenant lanes within one shard."""
import json
import time

import redis


class FairScheduler:
    """Deficit round-robin over per-tenant lanes, priced in worker-seconds."""

    def __init__(
        self,
        client: redis.Redis,
        shard: int,
        quantum_seconds: float = 2.0,
        max_in_flight_per_tenant: int = 8,
    ):
        self.r = client
        self.shard = shard
        self.ring = f"wh:ring:{shard}"            # LIST used as a rotating ring
        self.credits = f"wh:credits:{shard}"      # HASH tenant -> seconds owed
        self.weights = f"wh:weights:{shard}"      # HASH tenant -> relative weight
        self.quantum = quantum_seconds
        self.cap = max_in_flight_per_tenant

    def lane(self, tenant: str) -> str:
        return f"wh:lane:{self.shard}:{tenant}"

    def enqueue(self, envelope: dict) -> None:
        tenant = envelope["tenant"]
        pipe = self.r.pipeline()
        pipe.lpush(self.lane(tenant), json.dumps(envelope))
        pipe.execute()
        # Join the ring only if not already a member; LPOS is O(n) over active lanes.
        if self.r.lpos(self.ring, tenant) is None:
            self.r.rpush(self.ring, tenant)

    def refill(self) -> None:
        """Called on a fixed tick; each tenant's credit grows with its weight."""
        weights = {k.decode(): float(v) for k, v in self.r.hgetall(self.weights).items()}
        total = sum(weights.values()) or 1.0
        pipe = self.r.pipeline()
        for tenant, weight in weights.items():
            pipe.hincrbyfloat(self.credits, tenant, self.quantum * weight / total)
        pipe.execute()

    def next_envelope(self, max_scan: int = 64) -> dict | None:
        for _ in range(max_scan):
            raw_tenant = self.r.lmove(self.ring, self.ring, "LEFT", "RIGHT")
            if raw_tenant is None:
                return None                       # ring empty: shard is idle
            tenant = raw_tenant.decode()
            if int(self.r.get(f"wh:inflight:{self.shard}:{tenant}") or 0) >= self.cap:
                continue                          # tenant already at its concurrency cap
            if float(self.r.hget(self.credits, tenant) or 0.0) <= 0.0:
                continue                          # over budget this tick
            raw = self.r.rpop(self.lane(tenant))
            if raw is None:
                self.r.lrem(self.ring, 1, tenant)  # lane drained: leave the ring
                continue
            self.r.incr(f"wh:inflight:{self.shard}:{tenant}")
            return json.loads(raw)
        return None

    def release(self, tenant: str, elapsed_seconds: float) -> None:
        """Charge the tenant for the capacity it actually used, not per message."""
        pipe = self.r.pipeline()
        pipe.decr(f"wh:inflight:{self.shard}:{tenant}")
        pipe.hincrbyfloat(self.credits, tenant, -elapsed_seconds)
        pipe.execute()


def run(scheduler: FairScheduler, deliver) -> None:
    last_refill = 0.0
    while True:
        now = time.monotonic()
        if now - last_refill >= scheduler.quantum:
            scheduler.refill()
            last_refill = now
        envelope = scheduler.next_envelope()
        if envelope is None:
            time.sleep(0.05)
            continue
        started = time.monotonic()
        try:
            deliver(envelope)
        finally:
            scheduler.release(envelope["tenant"], time.monotonic() - started)

Two design points deserve emphasis. The per-tenant in-flight cap is doing at least as much work as the credit ledger: without it, a tenant whose endpoint takes 30 seconds to time out can hold every worker in the pool simultaneously while consuming almost no messages. Fair dequeuing does nothing if the workers are already blocked. The cap is also the natural place to integrate a per-endpoint breaker — when the breaker for a tenant’s endpoint is open, the fair share it would have used goes to everyone else instead of into a doomed request, which is the practical reason circuit breaker patterns and fair scheduling belong in the same component.

The cap’s absolute value follows from the pool: if the shard’s workers can sustain 400 concurrent attempts, a cap of 8 lets a single tenant take 2% of it, which is only meaningful once you know the total — see sizing worker pools for webhook dispatch for deriving that number.

Second, weights are a product decision, not a technical one. Uniform weights mean a customer sending ten events an hour gets the same delivery capacity as one sending a million, which is defensible for latency but wasteful in aggregate. Most platforms weight by plan tier with a floor, so no tenant can ever be starved to zero.

Step 4: Spill a sustained hot tenant onto a dedicated lane

Fair scheduling bounds the damage; it does not give the whale enough throughput to drain its own two-million-event import in a reasonable time, and it never provides a hard guarantee to the tenants who share a shard. For the small number of accounts that stay hot, promote them to a dedicated lane with its own worker allocation. Keep the number of dedicated lanes strictly bounded — eight is a good ceiling — because every one of them is a permanent operational object.

The controller that implements this runs on a slow tick — once a minute is plenty — and needs hysteresis in both directions, or a bursty tenant will flap between shard and lane every few minutes and drag its ordering guarantees along with it.

"""Promotion controller: move sustained hot tenants to dedicated lanes, and back."""
import time

import redis

R = redis.Redis.from_url("redis://localhost:6379/0")
LANES = "wh:dedicated"              # HASH tenant -> lane index
HOT_SINCE = "wh:hot_since"          # HASH tenant -> unix ts of first hot observation
MAX_LANES = 8
PROMOTE_SHARE = 0.30                # of its shard's consumed worker-seconds
DEMOTE_SHARE = 0.10                 # hysteresis gap prevents flapping
PROMOTE_AFTER = 900                 # seconds continuously hot
DEMOTE_AFTER = 3600                 # seconds continuously cold


def evaluate(tenant: str, share: float, now: float | None = None) -> str:
    now = now or time.time()
    dedicated = R.hget(LANES, tenant) is not None

    if share >= PROMOTE_SHARE and not dedicated:
        first_seen = R.hget(HOT_SINCE, tenant)
        if first_seen is None:
            R.hset(HOT_SINCE, tenant, now)
            return "watching"
        if now - float(first_seen) < PROMOTE_AFTER:
            return "watching"
        used = {int(v) for v in R.hvals(LANES)}
        free = next((index for index in range(MAX_LANES) if index not in used), None)
        if free is None:
            return "no_lane_available"          # evict the coldest lane, or shape at ingest
        R.hset(LANES, tenant, free)
        R.hset("wh:tenant:pin", tenant, f"lane-{free}")
        R.hdel(HOT_SINCE, tenant)
        return f"promoted:lane-{free}"

    if share < DEMOTE_SHARE and dedicated:
        cold_key = f"wh:cold_since:{tenant}"
        if not R.set(cold_key, now, nx=True, ex=DEMOTE_AFTER * 2):
            if now - float(R.get(cold_key)) >= DEMOTE_AFTER:
                R.hdel(LANES, tenant)
                R.hdel("wh:tenant:pin", tenant)
                R.delete(cold_key)
                return "demoted"
        return "cooling"

    R.hdel(HOT_SINCE, tenant)
    return "steady"

Note that the “no lane available” branch is not an error — it is the point where the answer stops being isolation and starts being admission control. A tenant that cannot be given a lane should be shaped at ingest instead, using the technique in applying backpressure to webhook consumers. Queues are a buffer, not a solution to a producer that is permanently faster than its consumers.

Step 5: Keep resharding safe for ordered events

Moving a tenant between shards is the one operation in this design that can violate ordering. If you write to lane B while lane A still holds unprocessed envelopes for the same tenant, two workers can process the same tenant’s events concurrently and deliver them out of order. Drain first, then move:

"""Drain-then-move: relocate a tenant's lane without concurrent processing."""
import json
import time

import redis

R = redis.Redis.from_url("redis://localhost:6379/0")


def relocate(tenant: str, target: str, timeout: float = 120.0) -> bool:
    """1) stop new enqueues to the old lane, 2) wait for it to drain, 3) repin."""
    R.hset("wh:paused", tenant, "1")           # producers buffer to wh:hold:{tenant}
    deadline = time.time() + timeout
    source = R.hget("wh:tenant:pin", tenant)
    source_lane = f"wh:lane:{source.decode()}:{tenant}" if source else None
    while time.time() < deadline:
        drained = source_lane is None or R.llen(source_lane) == 0
        quiet = int(R.get(f"wh:inflight:{tenant}") or 0) == 0
        if drained and quiet:
            break
        time.sleep(0.5)
    else:
        R.hdel("wh:paused", tenant)
        return False                            # never move a lane that is still busy
    R.hset("wh:tenant:pin", tenant, target)
    held = R.lrange(f"wh:hold:{tenant}", 0, -1)
    if held:
        pipe = R.pipeline()
        for raw in reversed(held):              # preserve original order
            pipe.lpush(f"wh:lane:{target}:{tenant}", raw)
        pipe.delete(f"wh:hold:{tenant}")
        pipe.execute()
    R.hdel("wh:paused", tenant)
    return True

The buffer window is bounded by the drain timeout, so pick a target that the lane can realistically reach — relocating a tenant with a 200,000-envelope backlog means waiting for that backlog, which is exactly when you least want to wait. Promote tenants before the backlog builds, on the trend rather than the peak.

Verification and testing

The property to assert is not “the whale gets less” but “the small tenant’s wait is independent of the whale’s backlog”. Simulate it:

"""Fairness regression test: a whale must not inflate a small tenant's wait."""
import fakeredis

from delivery.fair import FairScheduler


def test_small_tenant_is_not_blocked_by_a_whale():
    scheduler = FairScheduler(fakeredis.FakeStrictRedis(), shard=0, quantum_seconds=1.0)
    scheduler.r.hset(scheduler.weights, mapping={"whale": 1.0, "small": 1.0})

    for index in range(10_000):
        scheduler.enqueue({"tenant": "whale", "id": f"w{index}"})
    scheduler.enqueue({"tenant": "small", "id": "s0"})

    scheduler.refill()
    served, position = [], None
    for step in range(200):
        envelope = scheduler.next_envelope()
        if envelope is None:
            scheduler.refill()
            continue
        served.append(envelope["tenant"])
        scheduler.release(envelope["tenant"], 0.01)
        if envelope["id"] == "s0":
            position = step
            break

    assert position is not None and position < 20   # not behind 10,000 whale events
    assert served.count("small") == 1


def test_tenant_cannot_exceed_its_in_flight_cap():
    scheduler = FairScheduler(fakeredis.FakeStrictRedis(), shard=1, max_in_flight_per_tenant=3)
    scheduler.r.hset(scheduler.weights, mapping={"whale": 1.0})
    for index in range(50):
        scheduler.enqueue({"tenant": "whale", "id": index})
    scheduler.refill()

    leased = [scheduler.next_envelope() for _ in range(10)]
    assert sum(1 for envelope in leased if envelope is not None) == 3

In staging, run the real thing: push 100,000 envelopes for one tenant while a second tenant sends one event every ten seconds, and chart the second tenant’s queue wait. It should be flat. Then check the shard balance in production, because a hash that happens to concentrate your five largest customers on one shard is a configuration problem you want to find before an incident does:

# Per-shard depth: the spread matters more than any single value.
for shard in $(seq 0 31); do
  printf '%s\t%s\n' "$shard" "$(redis-cli llen "wh:shard:$shard")"
done | sort -k2 -nr | head

# Per-tenant share of consumed capacity, highest first.
redis-cli hgetall wh:tenant:seconds | paste - - | sort -k2 -nr | head

Failure modes and gotchas

• Partitioning on the wrong key. Hashing the event ID or the delivery ID spreads load beautifully and provides no isolation whatsoever, because a whale’s events land in every shard. It also destroys per-tenant ordering. The partition key must be the isolation boundary you actually care about — tenant, or subscription within a tenant if one customer runs many endpoints of wildly different quality.
• Fairness counted in messages instead of worker-seconds. A tenant whose endpoint always times out at 30 seconds consumes hundreds of times more capacity per message than one answering in 50 ms, yet a message-counting scheduler treats them as equal. Charge the credit ledger with elapsed delivery time, as release() does, or your fair scheduler will be systematically unfair to fast, well-behaved endpoints.
• No per-tenant concurrency cap. With fair dequeuing but unlimited concurrency, one tenant with slow endpoints still occupies every worker; the queue looks balanced while the pool is entirely consumed. The in-flight cap is what converts fair scheduling into fair capacity, and it must be enforced at lease time, not at enqueue.
• Resharding with modulo hashing. hash(tenant) % n remaps the overwhelming majority of tenants when n changes, so a capacity increase turns into a mass ordering violation and a stampede of half-drained lanes. Rendezvous or consistent hashing limits movement to the tenants that must move, and the drain-then-move procedure keeps even those safe.
• Dedicated lanes with no exit path. Promotion is easy to implement and easy to forget to reverse, so a year later you are running 60 dedicated lanes for tenants who churned or shrank, each with its own pollers and dashboards. Enforce the lane cap in code, require a demotion rule with hysteresis, and alert when the lane count sits at its ceiling — that is the signal that ingest shaping, not more lanes, is the answer.

Frequently Asked Questions

Related

• Webhook Delivery Queue Architecture — the topology these partitions live inside.
• Choosing a Message Broker for Webhook Delivery — which brokers support keyed partitioning natively.
• Sizing Worker Pools for Webhook Dispatch — how many workers each lane’s share is worth.
• Applying Backpressure to Webhook Consumers — what to do when isolation is no longer enough.