Backpressure-Aware Async Crawling: Coupling Your aiohttp Semaphore to Proxy Health

The standard asyncio scraping recipe is: spin up an aiohttp session, wrap each fetch in a asyncio.Semaphore(N), and crank N until throughput looks good or blocks start flying. It's a fine first step. It's also a blunt instrument, because it treats concurrency as a fixed setting while the network underneath it is anything but fixed. Your proxy pool's latency drifts, a region starts rate-limiting, one gateway IP goes stale — and a static semaphore blindly pushes the same 50 requests per second into a pipeline that just quietly degraded to 8.

The better model is backpressure: let the health of your own requests feed back into how many you allow in flight. This article shows a concrete pattern for tuning aiohttp concurrency against live signal rather than a magic number, without dropping requests or hammering origin servers when the pool turns sour.

Why a fixed semaphore is wrong under residential proxies

With a single IP and a polite target, a fixed concurrency cap is usually fine. With a rotating residential pool, each request may traverse a different gateway, different country, different upstream carrier. That means:

• Latency is a distribution, not a constant. The 20th request in a burst might route through a slower ASN.
• Effective capacity fluctuates. A provider hiccup can shrink usable pool size by half for minutes.
• 429s are sticky. Once a target starts limiting, pushing more concurrency makes the penalty worse, not better.

A fixed Semaphore(50) ignores all three. It keeps issuing 50 in-flight requests whether the pool is healthy or on fire. Backpressure closes the loop: measure, adjust the cap, repeat.

The core: an adaptive limiter

Rather than asyncio.Semaphore, implement a small limiter that adjusts its ceiling based on a rolling window of outcomes. The policy is classic additive-increase / multiplicative-decrease (AIMD), the same family of rules TCP congestion control uses.

import asyncio
import random
import time
from collections import deque


class AdaptiveLimiter:
    """Concurrency ceiling that grows on clean results and collapses on trouble."""

    def __init__(self, floor=2, ceiling=80, start=10, window=40):
        self.floor = floor
        self.ceiling = ceiling
        self.limit = start
        self.inflight = 0
        self.results = deque(maxlen=window)  # True=ok, False=trouble
        self._cond = asyncio.Condition()

    def _recompute(self):
        if len(self.results) < self.results.maxlen:
            return  # not enough signal yet
        bad = sum(1 for ok in self.results if not ok)
        ratio = bad / len(self.results)
        if ratio == 0.0:
            # clean window: grow toward ceiling, additive
            self.limit = min(self.ceiling, self.limit + 2)
        elif ratio < 0.05:
            self.limit = max(self.floor, int(self.limit * 1.05))
        elif ratio < 0.20:
            self.limit = max(self.floor, int(self.limit * 0.9))  # hold-ish
        else:
            # trouble: multiplicative decrease
            self.limit = max(self.floor, int(self.limit * 0.5))

    async def acquire(self):
        async with self._cond:
            while self.inflight >= self.limit:
                await self._cond.wait()
            self.inflight += 1

    async def release(self, ok: bool):
        async with self._cond:
            self.inflight -= 1
            self.results.append(ok)
            self._recompute()
            self._cond.notify()


class gated:
    """async context manager so we never leak a permit on exception."""

    def __init__(self, limiter: AdaptiveLimiter):
        self.lim = limiter

    async def __aenter__(self):
        await self.lim.acquire()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        # An exception is treated as trouble by default.
        await self.lim.release(ok=exc_type is None)
        return False

The key detail: release() records the outcome, and only after a full window of evidence does _recompute() move the ceiling. That prevents the limiter from thrashing on a single slow response.

Wiring it into aiohttp with proxy health in mind

Each request picks a gateway and reports whether the result was clean. "Clean" means: 2xx/3xx, no timeout, latency under your budget. Anything else is trouble and nudges the ceiling down.

import aiohttp
import asyncio


async def fetch(session, url, gateway, limiter, timeout_s=15):
    async with gated(limiter):
        t0 = time.perf_counter()
        try:
            async with session.get(
                url,
                proxy=gateway,
                timeout=aiohttp.ClientTimeout(total=timeout_s),
            ) as resp:
                await resp.read()
                dt = time.perf_counter() - t0
                ok = resp.status < 400 and dt < timeout_s * 0.8
                return resp.status, round(dt, 2), ok
        except (aiohttp.ClientError, asyncio.TimeoutError):
            return None, round(time.perf_counter() - t0, 2), False


async def worker(session, urls, gateways, limiter, results):
    for url in urls:
        # Rotate gateways; a real system would also skip known-bad ones.
        gw = random.choice(gateways)
        status, dt, ok = await fetch(session, url, gw, limiter, )
        results.append((status, dt, ok))


async def main(urls, gateways):
    limiter = AdaptiveLimiter(floor=2, ceiling=60, start=8, window=30)
    results = []
    connector = aiohttp.TCPConnector(limit=0, ssl=False)  # limiter owns the cap
    async with aiohttp.ClientSession(connector=connector) as session:
        shards = [urls[i::4] for i in range(4)]
        await asyncio.gather(
            *[worker(session, s, gateways, limiter, results) for s in shards]
        )
    ok = sum(1 for _, _, o in results if o)
    print(f"requests={len(results)} ok={ok} final_limit={limiter.limit}")
    return limiter.limit


if __name__ == "__main__":
    # Demo targets; swap for your real job + a provider gateway list.
    demo = [f"https://httpbin.org/status/{random.choice([200, 200, 200, 429, 503])}"
            for _ in range(300)]
    gateways = ["http://user-x:[e-mail]:8000",
                "http://user-x:[e-mail]:8000"]
    final = asyncio.run(main(demo, gateways))
    print("backpressure settled concurrency at:", final)

Run that against a mixed set of clean and failing responses and watch the ceiling climb while results are green, then clamp down the moment the failure ratio crosses your threshold. That is your throughput governor — no manual tuning, no dropped work.

What to measure per window

Don't gate on status code alone. A proxy that returns 200 after 14 seconds is quietly stealing your throughput even though nothing "failed." Feed the limiter a blended signal:

• Latency p95 vs your timeout budget — creeping latency is the earliest warning of pool degradation.
• 429/403 density by region — if only EU gateways are limited, your fix is to down-weight those gateways, not to cut global concurrency.
• Timeout rate — the most expensive failure, because it holds a permit open the longest.

In the code above, a slow-but-200 response already counts as not-ok via the dt < timeout_s * 0.8 term, so the limiter respects latency, not just errors.

Pitfalls that bite people

Permit leaks. Always release through the context manager's __aexit__; a raw try/finally around await limiter.acquire() will still leak if acquire itself is cancelled. The gated wrapper handles it.

Overshooting on recovery. Additive increase, multiplicative decrease is deliberate: grow slowly, fall fast. A limiter that doubles on one clean window will re-trigger the exact rate limit it just backed away from.

Forgetting the connector cap. If you set TCPConnector(limit=0) but never actually gate through the limiter, aiohttp will happily open hundreds of sockets. The limiter must be the sole governor of in-flight requests.

Global ceiling masking per-target limits. If you scrape five domains from one pipeline, a shared limiter lets a slow domain throttle the fast ones. Keep one limiter per target host if their politeness budgets differ.

The takeaway

Treat concurrency as a variable your program computes, not a constant you configure. An aiohttp crawler whose semaphore follows proxy health will finish roughly the same volume on good days and gracefully self-throttle on bad ones — instead of picking one of two failure modes: too slow, or blocked. Build the feedback loop once, and the number tunes itself forever after.

Disclosure: I use Thordata's residential proxies for this project. New users get 500MB free — code thor020 (10% off): https://www.thordata.com/?ls=uXcSHJzx&lk=02-tele