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
# ---------------------------------------------------------------------------