From dee689cd919f9ef2999cef773354adeb5e12f439 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Fri, 22 May 2026 06:55:32 -0400 Subject: [PATCH] fix+perf: fast-mode ignores crawl-delay; shared session; drop HEAD MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Full --fast crawl of russell.ballestrini.net (242 URIs): 26s -> ~5s. Three changes, biggest first: 1. fast_mode now actually ignores crawl-delay (the ~5x). The delay was only zeroed on the robots-200 path; a site with no robots.txt (404) or a robots fetch error fell back to default_crawl_delay (2s). Under --fast that made every concurrent fetch wave sleep ~2s — ~10 waves x 2s dominated the wall time. _enforce_crawl_delay now short-circuits when fast_mode, matching the documented "ignore crawl-delay" contract regardless of robots status. Disallow is still honored (separate path). 2. One shared ClientSession for the fetcher's lifetime (keepalive TCP connector sized to page-worker width) instead of a fresh session per fetch — ~3x on a 24-page wave. Lazily built in-loop via _get_session; the bridge closes it in a finally (guarded on owning the fetcher). 3. Drop the per-page preflight HEAD. aiohttp exposes response headers before the body is read, so the existing content-type binary guard skips images/video/audio without downloading them — the HEAD was a redundant round trip that doubled per-page latency. Diverges arborist's AsyncWebFetcher from the agents.ai.unturf.com/core verbatim lift (fox-approved); candidate to upstream. Regression tests pin fast=no-delay / polite=delay, shared-session lifecycle, and bridge session teardown (owned vs injected). --- arborist/sources/crawler/async_web_fetcher.py | 246 ++++++++++-------- arborist/sources/crawler/bridge.py | 110 ++++---- tests/crawler/test_async_web_fetcher.py | 70 +++++ tests/crawler/test_bridge.py | 51 ++++ 4 files changed, 321 insertions(+), 156 deletions(-) diff --git a/arborist/sources/crawler/async_web_fetcher.py b/arborist/sources/crawler/async_web_fetcher.py index 0c3ac48..234f3c2 100644 --- a/arborist/sources/crawler/async_web_fetcher.py +++ b/arborist/sources/crawler/async_web_fetcher.py @@ -1179,6 +1179,13 @@ class AsyncWebFetcher: self.domain_crawl_delays: Dict[str, float] = {} self.domain_last_fetched: Dict[str, float] = {} + # One shared ClientSession for the fetcher's lifetime, built + # lazily inside the event loop on first fetch (see _get_session). + # Reused across every page so a fast wide crawl keeps connections + # alive instead of paying TCP+TLS per request — the dominant cost + # once crawl-delay is 0. Released by aclose(). + self._session: Optional["aiohttp.ClientSession"] = None + # Page cache: {url: (html, links, timestamp)} self.page_cache: Dict[str, Tuple[str, List, float]] = {} @@ -1189,6 +1196,38 @@ class AsyncWebFetcher: logger.info(f"AsyncWebFetcher initialized with user-agent: {self.user_agent}") + async def _get_session(self) -> "aiohttp.ClientSession": + """Lazily build one shared ClientSession for the fetcher's life. + + A shared session (with a keepalive TCP connector sized to the + page-worker width) lets a wide fast crawl reuse connections + across pages instead of opening TCP+TLS per request — measured + ~3x on a 24-page wave. Must be built inside a running event loop; + release it with :meth:`aclose` when the crawl finishes. + """ + if self._session is None or self._session.closed: + connector = aiohttp.TCPConnector( + limit=max(self.max_page_workers * 2, 10), + limit_per_host=max(self.max_page_workers, 1), + ttl_dns_cache=300, + keepalive_timeout=30, + enable_cleanup_closed=True, + ) + self._session = aiohttp.ClientSession(connector=connector) + return self._session + + async def aclose(self) -> None: + """Close the shared session and its pooled connections. + + Safe to call repeatedly and a no-op if no session was built. The + crawl driver calls this in a ``finally`` so a fast crawl doesn't + leak sockets (or emit aiohttp's "Unclosed client session" + warning) on the way out. + """ + if self._session is not None and not self._session.closed: + await self._session.close() + self._session = None + def _get_domain(self, url: str) -> str: """Extract domain from URL""" parsed = Uri(url) @@ -1209,33 +1248,33 @@ class AsyncWebFetcher: parser.set_url(robots_url) try: - async with aiohttp.ClientSession() as session: - async with session.get( - robots_url, - headers={"User-Agent": self.user_agent}, - timeout=aiohttp.ClientTimeout(total=5) - ) as response: - if response.status == 200: - text = await response.text() - parser.parse(text.splitlines()) - self.robot_parsers[domain] = parser + session = await self._get_session() + async with session.get( + robots_url, + headers={"User-Agent": self.user_agent}, + timeout=aiohttp.ClientTimeout(total=5) + ) as response: + if response.status == 200: + text = await response.text() + parser.parse(text.splitlines()) + self.robot_parsers[domain] = parser - # Extract crawl delay (ignored in fast mode) - if self.fast_mode: - self.domain_crawl_delays[domain] = 0.0 - else: - delay = parser.crawl_delay(self.user_agent) - self.domain_crawl_delays[domain] = ( - delay if delay is not None else self.default_crawl_delay - ) - - logger.info(f"Parsed robots.txt for {domain}, crawl delay: {self.domain_crawl_delays[domain]}s") - return parser + # Extract crawl delay (ignored in fast mode) + if self.fast_mode: + self.domain_crawl_delays[domain] = 0.0 else: - logger.info(f"No robots.txt for {domain} (status {response.status})") - self.robot_parsers[domain] = None - self.domain_crawl_delays[domain] = self.default_crawl_delay - return None + delay = parser.crawl_delay(self.user_agent) + self.domain_crawl_delays[domain] = ( + delay if delay is not None else self.default_crawl_delay + ) + + logger.info(f"Parsed robots.txt for {domain}, crawl delay: {self.domain_crawl_delays[domain]}s") + return parser + else: + logger.info(f"No robots.txt for {domain} (status {response.status})") + self.robot_parsers[domain] = None + self.domain_crawl_delays[domain] = self.default_crawl_delay + return None except Exception as e: logger.warning(f"Could not fetch robots.txt for {domain}: {e}") self.robot_parsers[domain] = None @@ -1258,7 +1297,17 @@ class AsyncWebFetcher: return can_fetch async def _enforce_crawl_delay(self, domain: str): - """Enforce crawl delay for a domain""" + """Enforce crawl delay for a domain. + + fast_mode's contract is "ignore crawl-delay" — so short-circuit + here, NOT just on the robots-200 path. The per-domain delay is + only zeroed in _fetch_robots_txt when robots.txt returns 200; + on a 404/error the delay falls back to default_crawl_delay (2s), + which would otherwise make every concurrent fetch wave sleep ~2s + even under --fast (the dominant cost on a robots-less site). + """ + if self.fast_mode: + return delay = self.domain_crawl_delays.get(domain, self.default_crawl_delay) last_fetched = self.domain_last_fetched.get(domain, 0) elapsed = time.time() - last_fetched @@ -1738,97 +1787,78 @@ class AsyncWebFetcher: domain = self._get_domain(url) await self._enforce_crawl_delay(domain) - # Fetch the page + # Fetch the page. One GET, no preflight HEAD: aiohttp delivers + # the response headers before the body is read, so the + # content-type binary guard below skips images/video/audio + # without downloading them — the HEAD was a redundant round trip + # (it doubled per-page latency, the dominant cost of a wide fast + # crawl). The session is shared (see _get_session) so connections + # are kept alive and reused across the whole crawl. try: - async with aiohttp.ClientSession() as session: - # HEAD request first to check content type without downloading - try: - async with session.head( - url, - headers={"User-Agent": self.user_agent}, - timeout=aiohttp.ClientTimeout(total=10), - allow_redirects=True - ) as head_response: - head_content_type = head_response.headers.get('Content-Type', '').lower() - mime_type = head_content_type.split(';')[0].strip() - logger.info(f"HEAD {url} -> {head_response.status}, Content-Type: {mime_type}") + session = await self._get_session() + async with session.get( + url, + headers={"User-Agent": self.user_agent}, + timeout=aiohttp.ClientTimeout(total=15), + allow_redirects=True, + max_redirects=3 + ) as response: + if response.status != 200: + LAST_FETCH_ERROR = {'type': 'http_error', 'details': f'HTTP {response.status}', 'url': url, 'status_code': response.status} + logger.warning(f"Failed to fetch {url}: HTTP {response.status}") + return None, [] - # Skip images, video, audio - these need vision mode, not text fetch - if mime_type.startswith(('image/', 'video/', 'audio/')): - LAST_FETCH_ERROR = { - 'type': 'binary_content', - 'details': f'Content is {mime_type} - requires vision mode for images', - 'url': url, - 'content_type': mime_type - } - logger.info(f"Skipping {mime_type} content (not text-extractable): {url}") - return None, [] - except Exception as e: - # HEAD failed, continue with GET anyway - logger.debug(f"HEAD request failed for {url}: {e}, continuing with GET") + # Check content type for binary files (PDF, etc.) + content_type = response.headers.get('Content-Type', '').lower() + mime_type = content_type.split(';')[0].strip() - async with session.get( - url, - headers={"User-Agent": self.user_agent}, - timeout=aiohttp.ClientTimeout(total=15), - allow_redirects=True, - max_redirects=3 - ) as response: - if response.status != 200: - LAST_FETCH_ERROR = {'type': 'http_error', 'details': f'HTTP {response.status}', 'url': url, 'status_code': response.status} - logger.warning(f"Failed to fetch {url}: HTTP {response.status}") - return None, [] + # Skip images/video/audio before reading the body — these + # need vision mode, not text fetch. + if mime_type.startswith(('image/', 'video/', 'audio/')): + LAST_FETCH_ERROR = { + 'type': 'binary_content', + 'details': f'Content is {mime_type} - requires vision mode for images', + 'url': url, + 'content_type': mime_type + } + logger.info(f"Skipping {mime_type} content (not text-extractable): {url}") + return None, [] - # Check content type for binary files (PDF, etc.) - content_type = response.headers.get('Content-Type', '').lower() - mime_type = content_type.split(';')[0].strip() + is_pdf = 'application/pdf' in content_type or url.lower().endswith('.pdf') - # Double-check for binary content that HEAD might have missed - if mime_type.startswith(('image/', 'video/', 'audio/')): - LAST_FETCH_ERROR = { - 'type': 'binary_content', - 'details': f'Content is {mime_type} - requires vision mode for images', - 'url': url, - 'content_type': mime_type - } - logger.info(f"Skipping {mime_type} content (not text-extractable): {url}") - return None, [] - - is_pdf = 'application/pdf' in content_type or url.lower().endswith('.pdf') - - if is_pdf: - # Handle PDF: read as binary and extract text - pdf_bytes = await response.read() - logger.info(f"Fetched PDF {url} ({len(pdf_bytes)} bytes)") - pdf_text = extract_text_from_pdf(pdf_bytes) - if pdf_text: - # Wrap extracted text in minimal HTML for consistent processing - # Extract filename from URL for title - pdf_filename = url.split('/')[-1].split('?')[0] or "PDF Document" - html = f"{pdf_filename}
{pdf_text}
" - logger.info(f"Extracted {len(pdf_text)} chars of text from PDF") - else: - LAST_FETCH_ERROR = {'type': 'pdf_extraction_failed', 'details': 'Could not extract text from PDF', 'url': url} - logger.warning(f"Failed to extract text from PDF: {url}") - return None, [] + if is_pdf: + # Handle PDF: read as binary and extract text + pdf_bytes = await response.read() + logger.info(f"Fetched PDF {url} ({len(pdf_bytes)} bytes)") + pdf_text = extract_text_from_pdf(pdf_bytes) + if pdf_text: + # Wrap extracted text in minimal HTML for consistent processing + # Extract filename from URL for title + pdf_filename = url.split('/')[-1].split('?')[0] or "PDF Document" + html = f"{pdf_filename}
{pdf_text}
" + logger.info(f"Extracted {len(pdf_text)} chars of text from PDF") else: - html = await response.text() - logger.info(f"Successfully fetched {url} ({len(html)} bytes)") + LAST_FETCH_ERROR = {'type': 'pdf_extraction_failed', 'details': 'Could not extract text from PDF', 'url': url} + logger.warning(f"Failed to extract text from PDF: {url}") + return None, [] + else: + html = await response.text() + logger.info(f"Successfully fetched {url} ({len(html)} bytes)") - # Extract links if requested. lxml parsing releases the - # GIL, so running it in a thread genuinely parallelises - # across a wide fetch wave (html.parser would not). - links = [] - if extract_links: - links = await asyncio.to_thread( - extract_page_links, html, url, extract_anchor_text - ) - logger.info(f"Extracted {len(links)} links from {url}") + # Extract links if requested. lxml parsing releases the + # GIL, so running it in a thread genuinely parallelises + # across a wide fetch wave (html.parser would not). + links = [] + if extract_links: + links = await asyncio.to_thread( + extract_page_links, html, url, extract_anchor_text + ) + logger.info(f"Extracted {len(links)} links from {url}") - # Cache in memory for future fetches in this session - self.page_cache[url] = (html, links, time.time()) + # Cache in memory for future fetches in this session + self.page_cache[url] = (html, links, time.time()) - return html, links + return html, links except asyncio.TimeoutError: LAST_FETCH_ERROR = {'type': 'timeout', 'details': 'Connection timeout (>15 seconds)', 'url': url} diff --git a/arborist/sources/crawler/bridge.py b/arborist/sources/crawler/bridge.py index 2fb8101..dc5dc5e 100644 --- a/arborist/sources/crawler/bridge.py +++ b/arborist/sources/crawler/bridge.py @@ -135,6 +135,10 @@ async def _crawl_seed_async( aggressive fetching is acceptable (your own sites, dumps you've been authorized to mirror, etc.). """ + # Only tear down the shared session if we built the fetcher; an + # injected fetcher belongs to the caller (it may be reused across + # crawls), so we leave its lifecycle alone. + owns_fetcher = fetcher is None fetcher = fetcher or AsyncWebFetcher(user_agent=USER_AGENT, fast_mode=fast) # Fetch width = how many same-domain pages we pull concurrently per @@ -164,60 +168,70 @@ async def _crawl_seed_async( # 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): - # 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: + while queue and (max_pages == 0 or len(discovered) < max_pages): + # 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()) - # gather preserves order, so results[i] pairs with wave[i]. - results = await asyncio.gather(*[_fetch(url) for url, _ in wave]) + # gather preserves order, so results[i] pairs with wave[i]. + results = await asyncio.gather(*[_fetch(url) for url, _ in wave]) - for (url, depth), (html, links) in zip(wave, results): - if html is None: - continue + for (url, depth), (html, links) in zip(wave, results): + if html is None: + continue - discovered.append(url) - if progress is not None: - progress.tick(len(discovered), inserted=len(queue)) - if depth >= max_depth: - continue + discovered.append(url) + if progress is not None: + progress.tick(len(discovered), inserted=len(queue)) + if depth >= max_depth: + continue - 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 - # Don't crawl feed/sitemap URLs: they're crawl - # infrastructure that ingest_crawled discards anyway, and - # a big feed (e.g. a multi-MB atom.xml) is a slow wave - # straggler that gates everything fetched alongside it. - # Mark seen so a later parent can't re-enqueue it. - if _looks_like_feed_url(normalized): + 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 + # Don't crawl feed/sitemap URLs: they're crawl + # infrastructure that ingest_crawled discards anyway, + # and a big feed (e.g. a multi-MB atom.xml) is a slow + # wave straggler that gates everything fetched + # alongside it. Mark seen so a later parent can't + # re-enqueue it. + if _looks_like_feed_url(normalized): + seen.add(normalized) + continue seen.add(normalized) - continue - seen.add(normalized) - queue.append((normalized, depth + 1)) + queue.append((normalized, depth + 1)) - if progress is not None: - progress.done(len(discovered), inserted=len(queue)) - return discovered + if progress is not None: + progress.done(len(discovered), inserted=len(queue)) + return discovered + finally: + # Release the shared session we opened. Guard on owns_fetcher so + # we never close a caller-supplied fetcher, and on hasattr so + # stub fetchers in tests (no aclose) pass straight through. + if owns_fetcher: + aclose = getattr(fetcher, "aclose", None) + if aclose is not None: + await aclose() def crawl_seed( diff --git a/tests/crawler/test_async_web_fetcher.py b/tests/crawler/test_async_web_fetcher.py index 6a03e03..4811c3c 100644 --- a/tests/crawler/test_async_web_fetcher.py +++ b/tests/crawler/test_async_web_fetcher.py @@ -28,6 +28,76 @@ from arborist.sources.crawler.async_web_fetcher import ( ) +class TestSharedSession: + """Shared keepalive session lifecycle (fast-crawl connection reuse).""" + + def test_get_session_is_shared(self): + async def go(): + f = AsyncWebFetcher() + s1 = await f._get_session() + s2 = await f._get_session() + assert s1 is s2 + await f.aclose() + + asyncio.run(go()) + + def test_aclose_closes_and_resets(self): + async def go(): + f = AsyncWebFetcher() + s = await f._get_session() + await f.aclose() + assert s.closed + assert f._session is None + await f.aclose() # idempotent — must not raise + + asyncio.run(go()) + + def test_get_session_rebuilds_after_close(self): + async def go(): + f = AsyncWebFetcher() + s1 = await f._get_session() + await f.aclose() + s2 = await f._get_session() + assert s2 is not s1 + assert not s2.closed + await f.aclose() + + asyncio.run(go()) + + +class TestFastModeIgnoresCrawlDelay: + """fast_mode must ignore crawl-delay regardless of robots.txt status. + + Regression: the delay was only zeroed on the robots-200 path, so a + site with no robots.txt (404) fell back to default_crawl_delay (2s) + and every concurrent fetch wave slept ~2s even under --fast.""" + + def test_fast_mode_skips_delay_even_with_default_delay_set(self): + async def go(): + f = AsyncWebFetcher(fast_mode=True) + # Simulate the robots-404 fallback: a non-zero per-domain delay. + f.domain_crawl_delays["x.com"] = 2.0 + f.domain_last_fetched["x.com"] = __import__("time").time() + t = __import__("time").time() + await f._enforce_crawl_delay("x.com") + assert (__import__("time").time() - t) < 0.5, "fast_mode must not sleep" + + asyncio.run(go()) + + def test_polite_mode_still_enforces_delay(self): + async def go(): + import time as _t + + f = AsyncWebFetcher(fast_mode=False, default_crawl_delay=0.3) + f.domain_crawl_delays["x.com"] = 0.3 + f.domain_last_fetched["x.com"] = _t.time() + t = _t.time() + await f._enforce_crawl_delay("x.com") + assert (_t.time() - t) >= 0.25, "polite mode must still wait" + + asyncio.run(go()) + + class TestExtractPageLinks: """Link extraction for BFS (lxml fast path + BS4 fallback parity).""" diff --git a/tests/crawler/test_bridge.py b/tests/crawler/test_bridge.py index edf6d6b..2ccb622 100644 --- a/tests/crawler/test_bridge.py +++ b/tests/crawler/test_bridge.py @@ -297,6 +297,57 @@ def test_crawl_seed_polite_fetches_one_at_a_time(): assert fetcher.peak_in_flight == 1, "polite crawl must stay one-at-a-time" +def test_crawl_seed_closes_owned_fetcher(monkeypatch): + """When the bridge builds the fetcher, it tears down the shared + session in a finally — no leaked sockets after a fast crawl.""" + closed = {"count": 0} + + class _ClosableFetcher: + max_page_workers = 4 + + def __init__(self, **kwargs): + pass + + async def fetch_webpage(self, url, extract_links=False, **kw): + return "", [] + + async def aclose(self): + closed["count"] += 1 + + from arborist.sources.crawler import bridge + + monkeypatch.setattr(bridge, "AsyncWebFetcher", _ClosableFetcher) + bridge.crawl_seed("https://x.com/", max_depth=1, max_pages=1) + assert closed["count"] == 1, "bridge must close a fetcher it created" + + +def test_crawl_seed_leaves_injected_fetcher_open(): + """An injected fetcher belongs to the caller — the bridge must NOT + close its session (it may be reused across crawls).""" + import asyncio + + closed = {"count": 0} + + class _ClosableFetcher: + max_page_workers = 4 + + async def fetch_webpage(self, url, extract_links=False, **kw): + return "", [] + + async def aclose(self): + closed["count"] += 1 + + from arborist.sources.crawler import bridge + + fetcher = _ClosableFetcher() + asyncio.run( + bridge._crawl_seed_async( + "https://x.com/", max_depth=1, max_pages=1, fetcher=fetcher + ) + ) + assert closed["count"] == 0, "bridge must not close a caller-supplied fetcher" + + # --------------------------------------------------------------------------- # ingest_crawled: stores ETag + Last-Modified per document_root # ---------------------------------------------------------------------------