Scaling a Crawl Across Workers

When One Process Is Not Enough#

A single-process crawler is the right size for most jobs. It becomes the wrong size when the backlog reaches hundreds of thousands of URLs and three limits appear together: one machine caps your sockets, CPU for parsing, and memory; one process is one point of failure for a crawl that runs for days; and one deploy or restart interrupts everything in flight.

Distributing a crawl means separating two things that a simple script mixes together:

• State: what is left to do, what is being worked on right now, what is done, and how fast each target may be hit.
• Execution: fetching and parsing.

Move all state into a shared store, and workers become interchangeable and disposable. You can add ten, kill three, and redeploy all of them without losing work. Before building this, make sure the single-process version is solid: the caching and resume patterns are the foundation, and distribution only adds coordination on top of them.

Choosing the Shared Store#

This lesson uses Redis because its primitives map directly onto the concepts. The same ideas (atomic claim, lease, acknowledgement, recovery) apply to every option.

Atomic Claim With a Lease#

The core requirement is that each URL is handed to exactly one worker at a time, and that a URL claimed by a worker that crashes eventually returns to the queue. Redis's BLMOVE (the successor to the older BRPOPLPUSH) atomically moves an item from the todo list to a processing list, so two workers can never receive the same item. A sorted set records when each lease expires.

import time
import redis

r = redis.Redis(decode_responses=True)

TODO, PROCESSING, LEASES = "crawl:todo", "crawl:processing", "crawl:leases"
SEEN, DONE = "crawl:seen", "crawl:done"

def enqueue(url):
    # SADD returns 1 only for a new member: dedupe and push in one decision
    if r.sadd(SEEN, url):
        r.lpush(TODO, url)

def claim(lease_seconds=120):
    url = r.blmove(TODO, PROCESSING, timeout=5, src="RIGHT", dest="LEFT")
    if url:
        r.zadd(LEASES, {url: time.time() + lease_seconds})
    return url

def finish(url):
    pipe = r.pipeline()            # MULTI/EXEC: all three or none
    pipe.lrem(PROCESSING, 1, url)
    pipe.zrem(LEASES, url)
    pipe.sadd(DONE, url)
    pipe.execute()

def recover_expired():
    now = time.time()
    for url in r.zrangebyscore(LEASES, 0, now):
        if r.zrem(LEASES, url):    # only one recoverer wins each url
            r.lrem(PROCESSING, 1, url)
            r.rpush(TODO, url)

Why each piece is there:

• sadd before lpush. Every worker discovers links. With 40 workers finding the same category links, the set's atomic "was this new?" answer is what keeps the queue free of duplicates. Normalize URLs before this call, or duplicates slip in under different spellings.
• The lease is per item. A common mistake is calling EXPIRE on the processing list, which sets a timer on the whole list and deletes every in-flight item at once when it fires. A sorted set scored by expiry time gives each URL its own deadline.
• recover_expired is safe to run from every worker. ZREM returns 1 only for the caller that actually removed the entry, so two workers running recovery at once cannot both re-queue the same URL.
• Choose the lease longer than the slowest normal item (fetch timeout plus retries plus parsing). Too short, and slow pages get processed twice. Too long, and crashed work waits unnecessarily. Long-running items can extend their lease by calling zadd again as a heartbeat.

This design gives at-least-once processing: a worker that finishes the fetch and saves the record but dies before finish will have its URL processed again. Make the output write idempotent (an upsert on a stable key) and duplicates become harmless.

Retries and Dead Letters#

A URL that crashes a worker every time (a malformed page that breaks the parser, for example) will cycle through recovery forever. Count attempts and park repeat offenders:

def fail(url, error, max_attempts=4):
    attempts = r.hincrby("crawl:attempts", url, 1)
    pipe = r.pipeline()
    pipe.lrem(PROCESSING, 1, url)
    pipe.zrem(LEASES, url)
    if attempts >= max_attempts:
        pipe.hset("crawl:dead", url, error[:500])
    else:
        pipe.lpush(TODO, url)       # back of the line
    pipe.execute()

Review crawl:dead after each run. It is usually small and full of genuinely broken pages, parser bugs, or a new block page you have not seen before.

A Shared Rate Budget#

With one process, a per-host delay is local state. With twelve workers, twelve independent delays mean twelve times the intended rate on the target. The budget must live in the shared store. A fixed-window counter is simple and correct:

def acquire(host, max_per_minute, poll=0.5):
    while True:
        window = int(time.time() // 60)
        key = f"rate:{host}:{window}"
        pipe = r.pipeline()
        pipe.incr(key)
        pipe.expire(key, 120)       # tidy up old windows automatically
        count, _ = pipe.execute()
        if count <= max_per_minute:
            return
        time.sleep(poll)

Every worker calls acquire(host, 60) before each request to that host. Two caveats: a fixed window allows a burst at each window boundary (up to twice the limit across two adjacent seconds), and a waiting worker keeps incrementing the counter, which is harmless but inflates it. If the target needs smooth pacing, move to a token bucket implemented as a small Lua script so the check and update happen atomically on the server. For the politeness reasoning behind the numbers, see rate limiting and politeness.

A useful structural alternative is to partition by host: route each host to one worker (for example by hashing the hostname), so per-host pacing is local again and only the work assignment is shared.

The Worker Loop and Graceful Shutdown#

Orchestrators such as Kubernetes and systemd send SIGTERM and wait a grace period before killing a process. A worker that exits mid-item leaves its URL leased until the lease expires; one that finishes its current item and stops claiming leaves a clean queue.

import signal
import socket
import os

stopping = False
WORKER_ID = f"{socket.gethostname()}:{os.getpid()}"

def _stop(signum, frame):
    global stopping              # without this, the flag is a local variable
    stopping = True

signal.signal(signal.SIGTERM, _stop)
signal.signal(signal.SIGINT, _stop)

while not stopping:
    r.hset("crawl:workers", WORKER_ID, int(time.time()))   # heartbeat
    recover_expired()
    url = claim()
    if url is None:
        continue                  # blmove timed out; loop re-checks `stopping`
    try:
        acquire(host_of(url), max_per_minute=60)
        record, links = fetch_and_parse(url)
        save(record)              # idempotent upsert
        for link in links:
            enqueue(normalize(link))
        finish(url)
    except Exception as exc:
        fail(url, repr(exc))

r.hdel("crawl:workers", WORKER_ID)

The five-second timeout on claim matters: a blocking call with no timeout would never notice the stop flag. Make sure the grace period your orchestrator allows is longer than your slowest item.

Watching the Fleet#

A distributed crawl needs a dashboard, even a crude one. These numbers answer most questions:

print({
    "todo": r.llen(TODO),
    "processing": r.llen(PROCESSING),
    "done": r.scard(DONE),
    "dead": r.hlen("crawl:dead"),
    "workers_alive": sum(1 for ts in r.hvals("crawl:workers")
                         if time.time() - int(ts) < 60),
})

How to read them:

• todo growing while done is flat: workers are stuck (rate budget too tight, or all blocked).
• processing much larger than the worker count: workers are dying without finishing, and recovery is carrying the load.
• dead climbing quickly: a parser bug or a new block page. Stop and look before burning through the queue.

Also watch Redis memory. A seen set of tens of millions of long URLs uses gigabytes; storing a hash of each normalized URL (for example the first 16 bytes of SHA-256) instead of the URL itself cuts this substantially. Enable Redis persistence (AOF or snapshots) if losing the queue on a Redis restart is unacceptable.

Politeness Scales Too#

More workers means more capacity, not permission to use it. The target's tolerance is unchanged by how many machines you rent. Keep the shared budget as the hard ceiling, identify your crawler honestly, and honour robots.txt from every worker. Spreading requests across many IP addresses to stay under per-IP limits is a separate decision with its own ethical weight, discussed in proxies and rotation.

Practice#

Run three workers against a local test site with 5,000 pages. While they run, kill one with kill -9, then restart it. Confirm from the counters that the killed worker's in-flight URL returned to todo after its lease expired, that done ends at exactly 5,000, and that the combined request rate logged by the test server never exceeded the shared budget.