diff --git a/arborist/sources/crawler/bridge.py b/arborist/sources/crawler/bridge.py index c09396e..74b211f 100644 --- a/arborist/sources/crawler/bridge.py +++ b/arborist/sources/crawler/bridge.py @@ -127,54 +127,85 @@ async def _crawl_seed_async( ``fast=True`` flips the verbatim AsyncWebFetcher into fast_mode: 5s timeouts (vs 15-60s), CPU*3 parallel page workers, and the - target's robots.txt ``crawl-delay`` directive is ignored. Robots - Disallow is still honored. Use only against domains where + target's robots.txt ``crawl-delay`` directive is ignored. The BFS + fetches each wave with up to ``fetcher.max_page_workers`` requests + in flight (CPU*3 under fast_mode, 1 otherwise — so polite crawls + stay one-at-a-time and the crawl-delay still serialises them). + Robots Disallow is still honored. Use only against domains where aggressive fetching is acceptable (your own sites, dumps you've been authorized to mirror, etc.). """ fetcher = fetcher or AsyncWebFetcher(user_agent=USER_AGENT, fast_mode=fast) - seen: set[str] = set() + + # Fetch width = how many same-domain pages we pull concurrently per + # BFS wave. fast_mode sets the underlying fetcher's max_page_workers + # to CPU*3 with crawl-delay forced to 0 (`async_web_fetcher.py`), so + # a fast crawl fans the level out; polite mode keeps it at 1 so the + # per-page crawl-delay still serialises same-domain fetches (running + # them concurrently would only queue on the delay). A stub/injected + # fetcher without the attribute defaults to 1 — at width=1 the wave + # loop below is byte-for-byte the original one-at-a-time BFS. + width = max(1, getattr(fetcher, "max_page_workers", 1)) + + seed_norm = _normalize(seed_url) + seen: set[str] = {seed_norm} discovered: list[str] = [] - queue: deque[tuple[str, int]] = deque([(_normalize(seed_url), 0)]) + queue: deque[tuple[str, int]] = deque([(seed_norm, 0)]) + + async def _fetch(url: str): + # extract_links=True is REQUIRED for BFS — the verbatim + # AsyncWebFetcher defaults it to False (caller opts in for the + # link-graph use case). Without it BFS terminates at the seed. + try: + return await fetcher.fetch_webpage(url, extract_links=True) + except Exception: + return None, None # max_pages == 0 means "no cap" — depth is the only bound. Useful for # exhaustive same-domain crawls where the operator trusts depth to # bound the crawl naturally. while queue and (max_pages == 0 or len(discovered) < max_pages): - url, depth = queue.popleft() - if url in seen: - continue - seen.add(url) + # Pull a wave of up to `width` URLs off the front, never more + # than the remaining max_pages budget so a wide level can't + # overshoot the cap. width==1 makes each wave a single URL — + # identical to the original sequential pop. + wave: list[tuple[str, int]] = [] + while ( + queue + and len(wave) < width + and (max_pages == 0 or len(discovered) + len(wave) < max_pages) + ): + wave.append(queue.popleft()) - try: - # extract_links=True is REQUIRED for BFS — the verbatim - # AsyncWebFetcher defaults it to False (caller opts in for the - # link-graph use case). Without it BFS terminates at the seed. - html, links = await fetcher.fetch_webpage(url, extract_links=True) - except Exception: - continue - if html is None: - continue + # gather preserves order, so results[i] pairs with wave[i]. + results = await asyncio.gather(*[_fetch(url) for url, _ in wave]) - discovered.append(url) - if progress is not None: - progress.tick(len(discovered), inserted=len(queue)) - if depth >= max_depth: - continue + for (url, depth), (html, links) in zip(wave, results): + if html is None: + continue - for link in links or []: - link_url = link[0] if isinstance(link, tuple) else link - if not isinstance(link_url, str): + discovered.append(url) + if progress is not None: + progress.tick(len(discovered), inserted=len(queue)) + if depth >= max_depth: continue - absolute = urllib.parse.urljoin(url, link_url) - normalized = _normalize(absolute) - if normalized in seen: - continue - if not _same_domain(seed_url, normalized): - continue - if not normalized.startswith(("http://", "https://")): - continue - queue.append((normalized, depth + 1)) + + for link in links or []: + link_url = link[0] if isinstance(link, tuple) else link + if not isinstance(link_url, str): + continue + absolute = urllib.parse.urljoin(url, link_url) + normalized = _normalize(absolute) + # Dedup at enqueue (not pop) so a URL linked from two + # parents in the same wave is fetched exactly once. + if normalized in seen: + continue + if not _same_domain(seed_url, normalized): + continue + if not normalized.startswith(("http://", "https://")): + continue + seen.add(normalized) + queue.append((normalized, depth + 1)) if progress is not None: progress.done(len(discovered), inserted=len(queue)) @@ -194,7 +225,8 @@ def crawl_seed( ``max_pages=0`` means "no cap" — depth is the only bound. Pass ``progress`` for stderr heartbeats while BFS runs. ``fast=True`` enables AsyncWebFetcher's fast_mode (5s timeouts, CPU*3 workers, - ignores robots crawl-delay). + ignores robots crawl-delay) and the BFS fans each wave out to + CPU*3 concurrent fetches; polite mode stays one-at-a-time. """ return asyncio.run( _crawl_seed_async( diff --git a/tests/crawler/test_bridge.py b/tests/crawler/test_bridge.py index 8979db2..78bf621 100644 --- a/tests/crawler/test_bridge.py +++ b/tests/crawler/test_bridge.py @@ -217,6 +217,66 @@ def test_crawl_seed_passes_extract_links_true(): ) +class _ConcurrencyFetcher: + """Stub that records peak in-flight fetches so a test can prove the + BFS fans out (or doesn't). `max_page_workers` mimics what fast_mode + sets on the real AsyncWebFetcher; the bridge keys wave width off it. + """ + + def __init__(self, page_links: dict[str, list[str]], max_page_workers: int): + self.page_links = page_links + self.max_page_workers = max_page_workers + self._in_flight = 0 + self.peak_in_flight = 0 + + async def fetch_webpage(self, url: str, extract_links: bool = False, **kw): + import asyncio + + self._in_flight += 1 + self.peak_in_flight = max(self.peak_in_flight, self._in_flight) + try: + await asyncio.sleep(0.02) # hold the slot so overlap is observable + return "", self.page_links.get(url, []) + finally: + self._in_flight -= 1 + + +def test_crawl_seed_fast_fetches_a_wave_concurrently(): + """fast_mode sets max_page_workers>1 → a level of sibling pages is + fetched with multiple requests in flight, not one-at-a-time.""" + import asyncio + + from arborist.sources.crawler import bridge + + # Seed fans out to 6 leaf pages, all at depth 1 — one wave. + page_links = {"https://x.com/": [f"/p{i}" for i in range(6)]} + fetcher = _ConcurrencyFetcher(page_links, max_page_workers=6) + out = asyncio.run( + bridge._crawl_seed_async( + "https://x.com/", max_depth=1, max_pages=0, fetcher=fetcher + ) + ) + assert len(out) == 7 # seed + 6 leaves + assert fetcher.peak_in_flight > 1, "fast crawl must overlap fetches in a wave" + + +def test_crawl_seed_polite_fetches_one_at_a_time(): + """max_page_workers==1 (polite default) → strictly sequential, so the + per-page crawl-delay still serialises same-domain fetches.""" + import asyncio + + from arborist.sources.crawler import bridge + + page_links = {"https://x.com/": [f"/p{i}" for i in range(6)]} + fetcher = _ConcurrencyFetcher(page_links, max_page_workers=1) + asyncio.run( + bridge._crawl_seed_async( + "https://x.com/", max_depth=1, max_pages=0, fetcher=fetcher + ) + ) + assert fetcher.peak_in_flight == 1, "polite crawl must stay one-at-a-time" + + # --------------------------------------------------------------------------- # ingest_crawled: stores ETag + Last-Modified per document_root # ---------------------------------------------------------------------------