Implement a Finite Concurrent Web Crawler

Problem Statement

Design and implement an in-process concurrent web crawler engine, ConcurrentCrawler, capable of crawling a finite set of web pages starting from a list of seed URLs. The crawler must coordinate multiple concurrent workers, ensure per-host politeness, manage transactional URL states, handle network errors with classified retries, and reliably detect when the crawl has finished.

Requirements

1. URL State Machine & Deduplication: Each URL must transition through strictly tracked states: QUEUED, FETCHING, COMPLETED, or FAILED. A URL must never be fetched more than once successfully. If the crawler is initialized with existing storage state, any URL left in FETCHING from an interrupted crawl must be reset to QUEUED.
2. Politeness: Consecutive HTTP requests to the same host domain (e.g., example.com) must be spaced by at least politeness_delay_ms milliseconds. Requests to different hosts may proceed concurrently.
3. Lossless Discovery: Links extracted from completed pages must be enqueued if they have not been seen or completed previously.
4. Classified Retries: Transient fetch failures (HTTP 429, 503, network timeouts) can be retried up to 3 times before transitioning to FAILED. Non-transient errors (HTTP 404, invalid protocol) transition directly to FAILED without retries.
5. Termination Detection: The crawl terminates when either max_pages have transitioned to COMPLETED, or when the frontier is empty and zero workers are actively fetching.

Class Interface

from typing import List, Dict, Set
from dataclasses import dataclass

@dataclass
class FetchResult:
    status_code: int
    outlinks: List[str]
    is_transient_error: bool

class Fetcher:
    def fetch(self, url: str) -> FetchResult:
        """Simulates an HTTP fetch for a URL."""
        pass

class ConcurrentCrawler:
    def __init__(
        self,
        fetcher: Fetcher,
        num_workers: int,
        politeness_delay_ms: int,
        max_pages: int
    ):
        pass

    def crawl(self, seed_urls: List[str]) -> Dict[str, str]:
        """Runs the crawl to completion and returns a mapping of URL -> final state."""
        pass

Constraints

• 1 <= num_workers <= 32
• 0 <= politeness_delay_ms <= 1000
• 1 <= max_pages <= 10000
• URLs are valid ASCII strings containing a resolvable host.

Example

# Seeds: ["https://a.com/1", "https://b.com/1"]
# "https://a.com/1" links to ["https://a.com/2", "https://b.com/1"]
# Max pages: 5, politeness_delay_ms: 100

crawler = ConcurrentCrawler(fetcher, num_workers=2, politeness_delay_ms=100, max_pages=5)
results = crawler.crawl(["https://a.com/1", "https://b.com/1"])

# Output:
# {
#   "https://a.com/1": "COMPLETED",
#   "https://b.com/1": "COMPLETED",
#   "https://a.com/2": "COMPLETED"
# }
# Note: "https://b.com/1" was not fetched twice.