Partitioning a Crawl Across Machines Without Duplicate Fetches

A single machine can run a polite crawl of a few hundred hosts at a useful rate. Beyond that, the frontier outgrows the machine’s memory, the crawler’s uplink saturates, or the latency to distant servers makes serial fetching wasteful. The obvious response is to run several crawlers and split the work. The non-obvious part is that most ways of splitting the work break the politeness guarantee that the single-machine design spent so much effort establishing.

Splitting by URL, round-robin, puts consecutive URLs from one host on different machines, each with its own delay timer, and the host receives the sum of the machines’ rates. Splitting by discovery order has the same effect. Splitting by hash of the URL has the same effect unless the hash key is the host. Only a partition whose unit is the host, or the site (meaning all hosts under one registrable domain), keeps each host’s request stream under a single scheduler.

Why the host is the indivisible unit

The invariant the frontier maintains is a timestamp per host: the earliest moment the host may be contacted again. That invariant is a piece of state, and any design that duplicates the state across machines will violate it unless the machines coordinate on every request, which costs more than the crawl. The single-machine frontier holds that timestamp in its back-end queue, which is why the distributed design has to preserve the queue’s ownership rather than its contents.

With host partitioning, each host’s timestamp lives on exactly one machine, and that machine is the only one that ever contacts the host. No coordination is needed for correctness of the delay, because the delay is enforced by the sole owner. The crawler’s aggregate rate to a host is the owner’s rate, regardless of how many machines are in the cluster — the same property that a per-host delay gives a single crawler.

The site-level refinement matters for the same reason as in the single-machine case: several hostnames can resolve to the same server, and if example.com and www.example.com land on different partitions, the server sees two independent streams. Partitioning on the registrable domain rather than on the hostname costs some granularity, since a large site on its own infrastructure is then confined to one machine, but it removes the alias case entirely.

The cost of host partitioning is load skew. A cluster of n machines does not get n equal shares, because the web’s host sizes are heavy-tailed: a handful of sites account for a large fraction of the fetchable URLs. Hashing hosts uniformly still produces unbalanced partitions, and the imbalance is not correctable by adding machines, because the largest site cannot be split without violating politeness. A cluster sized for the average partition will have one machine over capacity and several idle.

Assignment, and what hashing buys

The assignment from host to machine must satisfy three properties at once: it must be computable from the host name alone, it must be stable when machines are added or removed, and it must be re-derivable after a failure without a central directory.

Modulo hashing satisfies the first and fails the second. Adding one machine to a cluster of n changes the assignment of nearly every host, which means every host’s queue moves, which means every host’s delay timer resets and its in-flight work is discarded. Consistent hashing, which maps both hosts and machines onto a ring and assigns each host to the next machine clockwise, changes only the assignments in the interval between the new machine and its predecessor — roughly 1/n of hosts when adding the n+1th machine.

import bisect, hashlib

def build_ring(machines, replicas=100):
    """Ring is computed once per membership change, then read-only."""
    ring = []
    for m in machines:
        for r in range(replicas):
            h = hashlib.sha1(f"{m}#{r}".encode()).digest()
            ring.append((int.from_bytes(h[:8], "big"), m))
    ring.sort()
    return [p for p, _ in ring], [m for _, m in ring]

def owner(host, points, machines):
    key = int.from_bytes(hashlib.sha1(host.encode()).digest()[:8], "big")
    return machines[bisect.bisect_left(points, key) % len(machines)]

The replicas are there because a single point per machine gives an uneven split even with a good hash; placing many virtual points per machine smooths the distribution at the cost of a larger ring to search. The ring is small enough to hold in memory on every machine, which is what makes the assignment derivable locally: a machine that discovers a new host computes the owner itself, with no lookup.

The residual skew is the heavy tail again. Consistent hashing balances the number of hosts per machine, not the number of URLs per machine, and the two are not proportional.

Moving a host without losing its state

When a host changes owner — because a machine was added, removed, or failed — its frontier state must move with it: the pending URLs, the delay timestamp, the validators, and the change-rate estimates. Moving the pending URLs is straightforward if they are stored per host in a durable structure. Moving the timing state is what determines whether the transition is polite.

If the new owner does not know the last fetch time, it will fetch immediately, and the host may have been contacted a moment earlier by the old owner. Carrying the timestamp in the host’s state record fixes that, and the record must be durable before the first fetch under the new ownership. A simpler approximation that is also defensible: on reassignment, treat the host as if it had just been fetched, and impose a full delay before the new owner’s first request. That costs at most one delay per moved host and requires no coordination.

The failure case is worse than the planned-move case, because the old owner may have died with requests in flight. The new owner cannot know whether those requests reached the server. Treating the host as freshly fetched — the conservative option — is the only safe choice, and it is why the recovery path is the same as the planned path with an extra delay.

Durability also bounds the loss. A crawler that checkpoints its per-host state periodically loses the work since the last checkpoint on a crash, and the loss is proportional to the checkpoint interval times the fetch rate. Checkpointing after every fetch is too expensive; checkpointing every few minutes means a crash costs a few minutes of fetched-and-forgotten pages, which is acceptable because the pages are refetchable and the frontier’s URLs are still there if the queue is persisted.

Duplicate suppression across partitions

Host partitioning removes duplicate fetches by construction for a single host, but duplicates still arise from three sources.

Cross-host aliases. The same content served under two registrable domains — mirrors, syndication, and country variants. Partitioning by domain splits them, and both partitions fetch the same bytes. Suppression has to happen at index time, comparing content digests across the whole crawl, which requires a shared store and is the reason a distributed crawl needs a global dedup layer even when the frontier is partitioned. The cheaper aliases — the same host under two names, or the same URL with a default port written out — are removed before partitioning by URL normalization, which is what keeps the alias problem down to the genuinely distinct domains.

Redirect targets. A redirect from host A to host B means partition A discovers a URL owned by partition B. The URL must be handed to B rather than fetched by A, or politeness for B is violated. The handoff is a message between partitions, and its cost is one round trip of coordination per redirect that crosses a boundary — cheap relative to the fetch, and necessary for correctness.

Re-discovery. Two partitions can both discover the same URL from different link sources, and both can enqueue it if the dedup check is local. A global set of seen URLs is the direct fix and the expensive one, because it is a shared write for every discovered URL. The common compromise is a partition-local dedup plus an assignment check: a URL’s owner is computed from its host, so a partition that discovers a URL it does not own sends it to the owner and never enqueues it locally. Duplicates within one partition are then caught by the local set, and duplicates across partitions cannot occur because only one partition ever enqueues a given URL.

What the cluster cannot do that one machine can

The single-machine frontier can make globally optimal decisions, because it sees every host and every priority. A partitioned crawl cannot: each machine schedules its own hosts and knows nothing about the load on the others. Priorities become local, and a high-priority URL discovered on an overloaded machine waits behind that machine’s queue while a low-priority URL elsewhere is fetched promptly.

Two partial remedies exist. The first is to bias assignment by priority, so that machines holding high-priority sites are the ones given extra capacity. That conflicts with consistent hashing’s stability, since priority changes over time and rebalancing on priority moves hosts frequently. The second is a global work queue for high-priority URLs, with machines pulling from it when their local queues are idle; this reintroduces coordination, but only for the small subset of URLs that justify it.

The larger limitation is that the cluster’s total rate is not the sum of its machines’ rates. DNS resolution, the shared dedup store, and the message bus between partitions all consume capacity, and the fraction grows with cluster size. Beyond some size, adding a machine adds more coordination traffic than fetch capacity, and the crawl’s throughput stops rising. Where that point falls depends on the ratio of coordination operations to fetches, which is why the design that sends a URL to its owner rather than fetching it locally is worth the extra message: it keeps coordination proportional to discovery rather than to fetching.

The last property the cluster gives up is a single view of freshness. Change rates are estimated per partition from that partition’s fetches, so a host that moves between machines carries its estimate only if the estimate is part of the moved state. A crawl that recomputes rates after every reassignment effectively restarts its freshness model whenever the cluster is resized, and the resulting schedule is worse than either the old or the new one alone.