Handling Slow Webhook Consumers Without Starving the Dispatcher

A single customer deployed a synchronous integration that writes to a legacy ERP inside their webhook handler. Their endpoint now takes 25 seconds to return 200 OK, they send you 40 events per second, and your delivery latency for every other customer has quietly tripled. Nothing is erroring. This page is about that failure, which sits under webhook timeout and connection management and is deliberately not solvable with the tools in setting connect and read timeouts alone.

The reason it is hard is that a slow consumer is not a broken one. It satisfies every health signal you have: 0% error rate, 100% delivery success, no timeouts if your read budget is generous, no circuit breaker activity. It is consuming a shared resource — worker slots — at 125 times the rate of a well-behaved endpoint, and resource consumption is not something error-rate telemetry can see.

Prerequisites

• Python 3.11+ with asyncio and httpx 0.27+, running a worker pool of known, fixed size.
• Per-endpoint delivery latency and status-code counters already flowing to your metrics backend.
• A queue whose messages can be requeued with a delay (SQS ChangeMessageVisibility, Celery retry(countdown=…), or a delayed-set in Redis).
• Redis 7 if you need the concurrency allowance shared across dispatcher processes rather than local to one.
• A written policy for how much of the fleet any one tenant may occupy — this is a product decision before it is an engineering one.

Step 1: Separate slow-but-successful from failing

Before adding machinery, verify which failure you have, because the two need opposite treatments. Chart worker occupancy — how much wall-clock time each worker spends inside a delivery, grouped by destination — rather than request counts.

The query that surfaces this is share-of-occupancy per host: sum by (host) (rate(delivery_duration_seconds_sum[5m])) / sum(rate(delivery_duration_seconds_sum[5m])). Alert when any single destination exceeds a threshold you choose from tenant count — with 500 active endpoints, one taking more than 15% of total worker-seconds is an anomaly regardless of how healthy it looks.

Step 2: Give every endpoint a bounded concurrency allowance

Bulkhead isolation means each destination gets its own fixed lane of in-flight slots. Exceeding the lane does not queue inside the dispatcher; it returns immediately so the worker can pick up other work.

Size each lane with Little’s Law rather than by feel. To sustain R deliveries per second to an endpoint whose p95 response time is L seconds you need R × L concurrent slots. Then apply a fairness ceiling: no lane may exceed a fixed share of the worker pool, and the slowest endpoints get the smallest lanes, not the largest.

The bottom row is the point. Granting 1,000 slots is impossible; granting 4 is a deliberate decision that the consumer’s backlog is their problem, not a fleet-wide outage.

import asyncio
from dataclasses import dataclass


class LaneFull(Exception):
    """Raised when an endpoint's bulkhead lane has no free slot."""


@dataclass
class Lane:
    semaphore: asyncio.Semaphore
    size: int


class Bulkhead:
    def __init__(self, default_size: int = 8, overrides: dict[str, int] | None = None):
        self._default = default_size
        self._overrides = overrides or {}
        self._lanes: dict[str, Lane] = {}

    def _lane(self, endpoint_id: str) -> Lane:
        lane = self._lanes.get(endpoint_id)
        if lane is None:
            size = self._overrides.get(endpoint_id, self._default)
            lane = Lane(semaphore=asyncio.Semaphore(size), size=size)
            self._lanes[endpoint_id] = lane
        return lane

    def in_use(self, endpoint_id: str) -> int:
        lane = self._lane(endpoint_id)
        return lane.size - lane.semaphore._value

    async def run(self, endpoint_id: str, coro_factory, *, wait: float = 0.5):
        lane = self._lane(endpoint_id)
        try:
            await asyncio.wait_for(lane.semaphore.acquire(), timeout=wait)
        except asyncio.TimeoutError:
            raise LaneFull(endpoint_id)
        try:
            return await coro_factory()
        finally:
            lane.semaphore.release()


BULKHEAD = Bulkhead(default_size=8, overrides={"ep_slow_erp": 4})

The wait=0.5 acquisition deadline is what distinguishes a bulkhead from a queue. Waiting indefinitely for a slot still holds the worker; waiting half a second and giving up hands the worker back almost immediately.

Step 3: Shed and requeue instead of queueing in memory

A shed delivery is not a failed delivery. It never left the process, no side effect occurred, and the correct response is to return it to the broker with a delay — not to increment a failure counter, not to feed the circuit breaker, and not to consume a retry attempt.

import random

from prometheus_client import Counter

SHED = Counter("webhook_deliveries_shed_total", "Rejected at the lane", ["endpoint"])
DELIVERED = Counter("webhook_deliveries_ok_total", "Delivered", ["endpoint"])


async def handle_message(message, client, bulkhead: Bulkhead) -> None:
    endpoint_id = message.endpoint_id

    async def _send():
        return await client.post(message.url, json=message.payload)

    try:
        response = await bulkhead.run(endpoint_id, _send, wait=0.5)
    except LaneFull:
        SHED.labels(endpoint=endpoint_id).inc()
        # Requeue with jittered delay. Does NOT count as a delivery attempt.
        delay = random.uniform(5.0, 20.0)
        await message.requeue(delay_seconds=delay, count_as_attempt=False)
        return

    if response.status_code < 300:
        DELIVERED.labels(endpoint=endpoint_id).inc()
        await message.ack()
    else:
        await message.retry(reason=f"http_{response.status_code}")

Two properties keep this stable. The requeue delay is jittered, because a fixed delay reconverges every shed message onto the same future instant and produces a periodic stampede against a lane that is still full. And count_as_attempt=False prevents a busy endpoint from exhausting its retry budget on shedding alone and landing in the dead-letter queue while perfectly healthy. This is admission control at the sender, complementary to the receiver-side signalling described in applying backpressure to webhook consumers.

Watch the shed rate as a first-class signal. Sustained shedding above roughly 20% of offered load for an endpoint means its lane is permanently undersized relative to what the tenant is sending, and the resolution is a conversation, not a config bump.

Step 4: Add a latency-triggered breaker for endpoints that never recover

Bulkheads bound the damage but never stop sending. For an endpoint that has been at 25 seconds for six hours, continuing to trickle deliveries into a growing backlog serves nobody. The control you need trips on latency, not error rate.

import time
from collections import deque


class LatencyBreaker:
    """Trips when p95 latency stays above a ceiling, regardless of status codes."""

    def __init__(self, ceiling_s: float = 15.0, window_s: float = 300.0,
                 min_samples: int = 20, cooldown_s: float = 600.0):
        self.ceiling_s = ceiling_s
        self.window_s = window_s
        self.min_samples = min_samples
        self.cooldown_s = cooldown_s
        self._samples: deque[tuple[float, float]] = deque()
        self._opened_at: float | None = None

    def record(self, duration_s: float) -> None:
        now = time.monotonic()
        self._samples.append((now, duration_s))
        cutoff = now - self.window_s
        while self._samples and self._samples[0][0] < cutoff:
            self._samples.popleft()

    def _p95(self) -> float:
        values = sorted(d for _, d in self._samples)
        if not values:
            return 0.0
        return values[min(len(values) - 1, int(len(values) * 0.95))]

    def allows(self) -> bool:
        now = time.monotonic()
        if self._opened_at is not None:
            if now - self._opened_at < self.cooldown_s:
                return False
            self._opened_at = None
            self._samples.clear()
            return True
        if len(self._samples) >= self.min_samples and self._p95() > self.ceiling_s:
            self._opened_at = now
            return False
        return True

Set ceiling_s from the endpoint’s own SLO, not a global constant — 15 seconds is aggressive for an ERP integration and absurdly generous for an edge function. Threshold selection for the error-rate breaker that guards the right-hand branch is covered in tuning circuit breaker thresholds for webhooks, and endpoints that never come back should end at auto-disabling failing webhook endpoints.

Step 5: Escalate to the consumer with evidence

Every control above degrades the slow consumer’s own delivery rate, so they must be told, with numbers, before they discover it as missing data. Expose per-endpoint p50/p95/p99 response time as measured by you, the current lane size, the shed rate, and the resulting sustainable throughput. The message that lands is arithmetic: “at your current p95 of 25 s and a lane of 4, you can receive 0.16 events per second; you are generating 40.”

Publish the same figures on a status endpoint the customer can poll, and include the lane size and effective rate in the delivery-failure webhook you send them. The remediation is always on their side — process asynchronously, return 202 Accepted immediately, do the ERP write out of band — and giving them the measurement makes that conversation short.

Verification and testing

Assert the two invariants that matter: a saturated lane sheds instead of blocking, and a slow lane does not affect a fast one.

import asyncio

import pytest


@pytest.mark.asyncio
async def test_lane_sheds_when_full():
    bulkhead = Bulkhead(default_size=2)

    async def slow():
        await asyncio.sleep(1.0)
        return "ok"

    running = [asyncio.create_task(bulkhead.run("ep_a", slow)) for _ in range(2)]
    await asyncio.sleep(0.05)
    with pytest.raises(LaneFull):
        await bulkhead.run("ep_a", slow, wait=0.1)
    assert bulkhead.in_use("ep_a") == 2
    await asyncio.gather(*running)
    assert bulkhead.in_use("ep_a") == 0


@pytest.mark.asyncio
async def test_slow_lane_does_not_block_fast_lane():
    bulkhead = Bulkhead(default_size=2)

    async def slow():
        await asyncio.sleep(2.0)

    async def fast():
        return "quick"

    blockers = [asyncio.create_task(bulkhead.run("ep_slow", slow)) for _ in range(2)]
    await asyncio.sleep(0.05)
    started = asyncio.get_running_loop().time()
    assert await bulkhead.run("ep_fast", fast, wait=0.1) == "quick"
    assert asyncio.get_running_loop().time() - started < 0.1
    await asyncio.gather(*blockers)

The second test is the regression guard for the whole page: if someone replaces the per-endpoint semaphore with a single global one, it fails immediately.

For a live check, confirm shedding is actually happening and is bounded:

# Shed rate should be non-zero for the slow endpoint and zero everywhere else.
curl -s localhost:9090/metrics | grep webhook_deliveries_shed_total

Failure modes and gotchas

• Raising the read timeout instead of capping concurrency. The instinctive response to a 25-second endpoint is to raise the read budget to 30 s. That does not reduce occupancy at all — it guarantees it, by ensuring every slow delivery runs to completion while holding a worker. Cap the lane first; adjust the timeout only after concurrency is bounded.
• An unbounded in-memory wait queue in front of the lane. Replacing wait=0.5 with an unbounded await semaphore.acquire() looks like it preserves ordering and loses nothing. It converts the bulkhead into a parking lot: workers block on the semaphore, occupancy is unchanged, and memory grows with the backlog. The short acquisition deadline is the mechanism, not a detail.
• Counting shed events as delivery failures. Feeding shed outcomes into the error-rate breaker or the retry counter drives a healthy-but-throttled endpoint straight into auto-disable and the dead-letter queue. Shed is a third outcome alongside success and failure, and it must be carried as one all the way through the retry layer.
• Per-process lanes on a multi-process fleet. A local semaphore of 4 across 12 dispatcher pods is a fleet-wide allowance of 48. Either divide the intended allowance by replica count and accept the drift, or move the counter into Redis with a TTL-guarded INCR/DECR so the lane is genuinely global.

Frequently Asked Questions

Related

• Setting connect and read timeouts
• Connection pooling and keep-alive for webhook senders
• Applying backpressure to webhook consumers
• Webhook timeout and connection management