Politeness-Aware Frontier Scheduling With Celery #

A task queue distributes work; it does not know that two tasks target the same host and must not run within two seconds of each other. Bolting per-host pacing onto a general-purpose queue naively produces workers that sleep on host timers while thousands of URLs for healthy hosts wait behind them. This guide builds a frontier that keeps a task queue busy without ever blocking on politeness, as part of distributed crawl scheduling and queues in the Network Resilience & Proxy Management section.

Problem Framing #

The tempting design is one task per URL, with the worker sleeping until the host’s limiter permits the request. It fails at moderate scale for a specific reason: workers are a fixed resource and sleeping consumes one. A crawl of ten thousand URLs across fifty hosts with a two-second interval needs, at steady state, twenty-five concurrent requests — but a naive design occupies as many workers as there are queued tasks, nearly all of them asleep.

Eligibility is decided before a task is enqueued, never inside it.

The second tempting design is a per-host queue with a dedicated worker. It removes the sleeping but scales badly: a crawl touching a thousand hosts needs a thousand queues, most of them idle.

The design that works separates eligibility from execution. A frontier structure holds URLs per host and knows when each host may next be fetched; a small dispatcher polls it for eligible work and enqueues only that; workers execute immediately and never wait. Celery becomes the execution layer, and the politeness logic lives where it can see the whole picture.

Step-by-Step Implementation #

1. The frontier: a ready-set plus per-host lists #

A single sorted-set query finds an eligible host regardless of frontier size.

import time

DISPATCH_LUA = """
-- KEYS[1] = ready zset (member=host, score=earliest next dispatch time)
-- KEYS[2] = per-host list key prefix
-- ARGV[1] = now, ARGV[2] = batch size
local out = {}
for _ = 1, tonumber(ARGV[2]) do
  local hosts = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', ARGV[1], 'LIMIT', 0, 1)
  if #hosts == 0 then break end
  local host = hosts[1]
  local url = redis.call('LPOP', KEYS[2] .. host)
  if not url then
    redis.call('ZREM', KEYS[1], host)          -- host drained
  else
    table.insert(out, host)
    table.insert(out, url)
    redis.call('ZADD', KEYS[1], 'XX', ARGV[1], host)   -- placeholder; caller re-scores
  end
end
return out
"""

class Frontier:
    READY = "frontier:ready"
    PREFIX = "frontier:urls:"

    def __init__(self, redis):
        self.redis = redis
        self._dispatch = redis.register_script(DISPATCH_LUA)

    def add(self, host: str, url: str, eligible_at: float | None = None) -> None:
        pipe = self.redis.pipeline()
        pipe.rpush(f"{self.PREFIX}{host}", url)
        pipe.zadd(self.READY, {host: eligible_at or time.time()}, nx=True)
        pipe.execute()

    def take(self, batch: int = 50) -> list[tuple[str, str]]:
        flat = self._dispatch(keys=[self.READY, self.PREFIX], args=[time.time(), batch])
        return [(flat[i].decode(), flat[i + 1].decode()) for i in range(0, len(flat), 2)]

    def defer(self, host: str, until: float) -> None:
        self.redis.zadd(self.READY, {host: until})

A single sorted-set range query finds an eligible host in logarithmic time no matter how many hosts are deferred, which is what lets the dispatcher stay cheap as the crawl grows.

2. The dispatcher: the only component that knows about pacing #

from celery import Celery

app = Celery("crawler", broker=BROKER_URL)

DISPATCH_INTERVAL = 0.25
MAX_IN_FLIGHT = 200

def dispatch_loop(frontier: Frontier, registry, limiter, breakers) -> None:
    while True:
        if in_flight_count() >= MAX_IN_FLIGHT:
            time.sleep(DISPATCH_INTERVAL)
            continue

        for host, url in frontier.take(batch=50):
            if not breakers.allow(host):
                frontier.add(host, url, eligible_at=breakers.retry_at(host))
                continue

            wait = limiter.acquire(host)
            if wait > 0:
                # Not eligible yet: put it back and defer the WHOLE host.
                frontier.add(host, url, eligible_at=time.time() + wait)
                frontier.defer(host, time.time() + wait)
                continue

            fetch_url.apply_async(args=[host, url], queue="crawl")
            frontier.defer(host, time.time() + registry.interval(host))

        time.sleep(DISPATCH_INTERVAL)

Two properties make this work. The dispatcher never sleeps on a host — when a host is not eligible it defers the host and moves to the next one. And a task is enqueued only when it may execute immediately, so a Celery worker picking it up never waits.

Run one dispatcher, or a small number with a leader lock. It is not the bottleneck: a single process comfortably dispatches thousands of tasks per second, because all it does is a few Redis operations per URL.

3. The worker task: fetch, then feed results back #

@app.task(bind=True, acks_late=True, max_retries=0, time_limit=120, soft_time_limit=90)
def fetch_url(self, host: str, url: str) -> None:
    """Executes immediately — all pacing decisions were made by the dispatcher."""
    outcome, response = transport.fetch(url, host=host)

    limiter.observe(host, outcome, response.elapsed if response else None)
    breakers.record(host, outcome)

    if outcome == "throttled":
        frontier.add(host, url, eligible_at=time.time() + retry_after(response))
        frontier.defer(host, time.time() + retry_after(response))
        return
    if outcome == "transient" and self.request.retries < 3:
        frontier.add(host, url, eligible_at=time.time() + backoff(self.request.retries))
        return
    if outcome == "refused":
        breakers.force_open(host)
        governance.open_case(host=host, url=url, kind="identity_refusal")
        return

    for discovered in extract_and_store(url, response):
        if rules.can_fetch(host_of(discovered), discovered) and seen.add(discovered):
            frontier.add(host_of(discovered), discovered)

max_retries=0 is deliberate: retries go back through the frontier so they are subject to the same pacing as any other URL. A Celery-level retry bypasses the limiter entirely, which is exactly the hole this design exists to close.

acks_late=True with a time_limit gives crash safety — a worker that dies without acknowledging returns its task to the broker — while the time limit bounds how long a stuck fetch can hold it.

4. Bound per-host queue depth #

Discovery can enqueue faster than a politely paced host can be fetched, and an unbounded per-host list grows without limit.

MAX_PER_HOST = 50_000

def add_bounded(frontier: Frontier, host: str, url: str) -> bool:
    depth = frontier.redis.llen(f"{Frontier.PREFIX}{host}")
    if depth >= MAX_PER_HOST:
        metrics.frontier_rejected.labels(host=host).inc()
        return False
    frontier.add(host, url)
    return True

Rejecting rather than accepting is the right default. A host whose queue has reached fifty thousand URLs at a two-second interval represents nearly thirty days of crawling, and quietly accepting more is pretending to a plan that does not exist. Export the rejection count and the per-host depth, because a growing queue is the earliest signal that discovery is outrunning the budget.

Verification & Testing #

A broker-level retry bypasses the limiter entirely.

def test_dispatcher_never_blocks_on_a_slow_host(frontier, limiter):
    frontier.add("slow.example", "https://slow.example/a")
    for i in range(20):
        frontier.add("fast.example", f"https://fast.example/{i}")
    limiter.set_interval("slow.example", 3600.0)

    dispatched = run_dispatcher_once(frontier, limiter)
    assert all(host == "fast.example" for host, _ in dispatched)
    assert len(dispatched) == 20

def test_per_host_interval_holds_across_workers(simulator):
    result = simulator(workers=8, host="example.com", interval=2.0, urls=30, seconds=60)
    gaps = [b - a for a, b in zip(result.dispatch_times, result.dispatch_times[1:])]
    assert min(gaps) >= 1.99          # eight workers, still one request every 2 s

def test_crashed_worker_returns_its_url(broker, frontier):
    kill_worker_mid_task(broker)
    assert frontier.depth("example.com") == 1

The second test is the one worth keeping permanently: it asserts the property that distinguishes a politeness-aware frontier from a task queue with a sleep in it.

Compliance & Operational Guardrails #

All pacing decisions happen in the dispatcher; workers never sleep on a host timer.
Retries re-enter the frontier so they are paced like any other URL.
Disallowed URLs are filtered at insert time, never at dispatch time.
Per-host queue depth is bounded and its rejections are counted.
A refusal opens the host’s circuit and raises a case; it never re-queues the URL.

Common Mistakes #

Sleeping inside the worker task. Workers become a scarce resource consumed by waiting, and throughput collapses to the pace of the slowest host.
Using Celery’s own retry for crawl failures. Retries bypass the limiter and the breaker entirely.
Unbounded per-host queues. Discovery outruns the budget and the frontier grows until the store runs out of memory.

Frequently Asked Questions #

Why not one Celery queue per host? #

Because host count is unbounded and queue count is not free — brokers degrade with thousands of queues, and most of them would be empty. The ready-set indirection gives the same per-host isolation with one queue and one sorted set.

How many dispatchers should run? #

One active, with a leader lock and a warm standby. The dispatcher is not throughput-limited, and multiple active dispatchers race on the same eligible hosts, which is correct but wasteful. A lock with a short TTL gives failover within seconds.

Does this work with other task queues? #

Yes — nothing in the frontier depends on Celery. The same dispatcher works with RQ, Dramatiq, SQS, or a plain worker pool, because the only thing the queue provides is execution. That separation is the point: politeness logic in the frontier, execution anywhere.

How should the frontier be drained at the end of a crawl? #

By letting the dispatcher run until the ready set is empty and the in-flight count reaches zero, rather than by stopping it on a timer. A crawl stopped on a timer leaves URLs in per-host lists and in-flight lists, and the next run either reprocesses them or, worse, treats them as already seen. Draining explicitly and recording the final depths gives a clean boundary between generations.

What happens if the dispatcher falls behind? #

The symptom is workers idling while the frontier holds eligible URLs, and the usual cause is the dispatch batch being too small relative to the worker count. Raise the batch size before adding dispatchers: a single process handling a few thousand dispatches per second is rarely the constraint, and multiple active dispatchers race on the same eligible hosts. If a second is genuinely needed, partition the host space between them rather than letting both draw from the whole ready set.

Distributed Crawl Scheduling and Queues — the frontier invariants this implements.
Deduplicating URLs With a Redis Seen Set — the filter applied before a URL enters the frontier.
Implementing Polite Rate Limiting — the limiter the dispatcher consults for eligibility.