fix(deep_crawling): BFS O(n²) parent lookup and BestFirst duplicate enqueue

crawl4ai/deep_crawling/bff_strategy.py

@@ -71,6 +71,10 @@ def __init__(
         self._last_state: Optional[Dict[str, Any]] = None
         # Shadow list for queue items (only used when on_state_change is set)
         self._queue_shadow: Optional[List[Tuple[float, int, str, Optional[str]]]] = None
+        # URLs already scored/enqueued this crawl. Kept separate from `visited`
+        # (which tracks dequeued URLs) so a URL discovered by two sibling pages
+        # in the same batch is only pushed onto the priority queue once.
+        self._enqueued: Set[str] = set()
 
     async def can_process_url(self, url: str, depth: int) -> bool:
         """
@@ -177,12 +181,16 @@ async def link_discovery(
         for link in links:
             url = link.get("href")
             base_url = normalize_url_for_deep_crawl(url, source_url)
-            if base_url in visited:
+            if base_url in visited or base_url in self._enqueued:
                 continue
             if not await self.can_process_url(base_url, new_depth):
                 self.stats.urls_skipped += 1
                 continue
-
+
+            # Mark as enqueued now (not just when it's later put on the queue)
+            # so a second inbound link to the same URL, discovered from a
+            # sibling page later in this batch, is skipped here too.
+            self._enqueued.add(base_url)
             valid_links.append(base_url)
 
         # Record the new depths and add to next_links
@@ -216,6 +224,9 @@ async def _arun_best_first(
             queue_items = self._resume_state.get("queue_items", [])
             for item in queue_items:
                 await queue.put((item["score"], item["depth"], item["url"], item["parent_url"]))
+            # Already-visited and already-queued URLs are both settled; new
+            # discoveries should skip them.
+            self._enqueued = visited | {item["url"] for item in queue_items}
             # Initialize shadow list if callback is set
             if self._on_state_change:
                 self._queue_shadow = [
@@ -228,6 +239,7 @@ async def _arun_best_first(
             await queue.put((-initial_score, 0, start_url, None))
             visited: Set[str] = set()
             depths: Dict[str, int] = {start_url: 0}
+            self._enqueued = {start_url}
             # Initialize shadow list if callback is set
             if self._on_state_change:
                 self._queue_shadow = [(-initial_score, 0, start_url, None)]

crawl4ai/deep_crawling/bfs_strategy.py

@@ -248,6 +248,7 @@ async def _arun_batch(
 
             next_level: List[Tuple[str, Optional[str]]] = []
             urls = [url for url, _ in current_level]
+            parent_by_url: Dict[str, Optional[str]] = dict(current_level)
 
             # Clone the config to disable deep crawling recursion and enforce batch mode.
             batch_config = config.clone(deep_crawl_strategy=None, stream=False)
@@ -258,7 +259,7 @@ async def _arun_batch(
                 depth = depths.get(url, 0)
                 result.metadata = result.metadata or {}
                 result.metadata["depth"] = depth
-                parent_url = next((parent for (u, parent) in current_level if u == url), None)
+                parent_url = parent_by_url.get(url)
                 result.metadata["parent_url"] = parent_url
                 results.append(result)
 
@@ -336,19 +337,20 @@ async def _arun_stream(
 
             next_level: List[Tuple[str, Optional[str]]] = []
             urls = [url for url, _ in current_level]
+            parent_by_url: Dict[str, Optional[str]] = dict(current_level)
             visited.update(urls)
 
             stream_config = config.clone(deep_crawl_strategy=None, stream=True)
             stream_gen = await crawler.arun_many(urls=urls, config=stream_config)
-            
+
             # Keep track of processed results for this batch
             results_count = 0
             async for result in stream_gen:
                 url = result.url
                 depth = depths.get(url, 0)
                 result.metadata = result.metadata or {}
                 result.metadata["depth"] = depth
-                parent_url = next((parent for (u, parent) in current_level if u == url), None)
+                parent_url = parent_by_url.get(url)
                 result.metadata["parent_url"] = parent_url
 
                 # Count only successful crawls

tests/deep_crawling/test_deep_crawl_efficiency.py

@@ -0,0 +1,157 @@
+"""
+Regression tests for GH issue #2242:
+
+1. BFSDeepCrawlStrategy re-scanned the entire current level (a Python list)
+   once per fetched result to find that result's parent URL, making the
+   per-level bookkeeping O(n^2) instead of O(n).
+2. BestFirstCrawlingStrategy.link_discovery only checked `visited` without
+   updating it, so a URL discovered by two sibling pages in the same batch
+   was scored and enqueued twice instead of once.
+"""
+
+import asyncio
+import time
+
+import pytest
+from unittest.mock import MagicMock
+
+from crawl4ai.deep_crawling import BFSDeepCrawlStrategy, BestFirstCrawlingStrategy
+
+
+def create_mock_config(stream=False):
+    config = MagicMock()
+    config.stream = stream
+
+    def clone_config(**kwargs):
+        new_config = MagicMock()
+        new_config.stream = kwargs.get("stream", stream)
+        new_config.clone = MagicMock(side_effect=clone_config)
+        return new_config
+
+    config.clone = MagicMock(side_effect=clone_config)
+    return config
+
+
+def create_fanout_crawler(start_url, num_children):
+    """Mock crawler where `start_url` links to `num_children` leaf pages."""
+
+    async def mock_arun_many(urls, config):
+        results = []
+        for url in urls:
+            result = MagicMock()
+            result.url = url
+            result.success = True
+            result.metadata = {}
+            if url == start_url:
+                links = [
+                    {"href": f"{start_url}/child{i}"} for i in range(num_children)
+                ]
+            else:
+                links = []
+            result.links = {"internal": links, "external": []}
+            results.append(result)
+        return results
+
+    crawler = MagicMock()
+    crawler.arun_many = mock_arun_many
+    return crawler
+
+
+class TestBFSParentLookupPerformance:
+    """
+    The parent lookup used to be `next((parent for (u, parent) in
+    current_level if u == url), None)`, executed once per result in the
+    level: O(n) work per result, O(n^2) for the whole level. Measure the
+    strategy's own wall-clock growth between a small and a large level -
+    a self-relative check that doesn't depend on absolute machine speed.
+    Linear behavior keeps the ratio close to the level-size ratio (8x);
+    quadratic behavior pushes it towards its square (64x).
+    """
+
+    @pytest.mark.asyncio
+    async def test_parent_lookup_scales_linearly_with_level_size(self):
+        start_url = "https://example.com/start"
+        small_n, large_n = 2500, 20000
+
+        async def timed_run(num_children):
+            strategy = BFSDeepCrawlStrategy(max_depth=2)
+            crawler = create_fanout_crawler(start_url, num_children)
+            config = create_mock_config(stream=False)
+            t0 = time.perf_counter()
+            results = await strategy._arun_batch(start_url, crawler, config)
+            elapsed = time.perf_counter() - t0
+            assert len(results) == num_children + 1
+            return elapsed
+
+        small_time = await timed_run(small_n)
+        large_time = await timed_run(large_n)
+
+        growth = large_time / small_time
+        # Level size grows 8x; O(n) bookkeeping keeps growth near 8x while
+        # O(n^2) bookkeeping pushes it towards 64x. 12x cleanly separates
+        # the two on this workload (measured ~8.3x fixed, ~16.9x unfixed).
+        assert growth < 12, (
+            f"per-level bookkeeping does not scale linearly: {small_n} "
+            f"URLs took {small_time:.3f}s, {large_n} URLs took "
+            f"{large_time:.3f}s ({growth:.1f}x for an {large_n / small_n:.0f}x "
+            "increase in level size)"
+        )
+
+
+class CountingScorer:
+    """Records how many times each URL is scored."""
+
+    def __init__(self):
+        self.call_counts = {}
+
+    def score(self, url):
+        self.call_counts[url] = self.call_counts.get(url, 0) + 1
+        return 0.0
+
+
+class TestBestFirstDuplicateEnqueue:
+    """
+    `link_discovery` checked `if base_url in visited: continue` but never
+    added newly discovered URLs to `visited`. When two sibling pages in the
+    same batch link to the same third URL, that URL was scored and pushed
+    onto the priority queue twice before either copy was ever dequeued.
+    """
+
+    @pytest.mark.asyncio
+    async def test_shared_link_is_scored_and_enqueued_once(self):
+        start_url = "https://example.com/start"
+        page_a = "https://example.com/a"
+        page_b = "https://example.com/b"
+        shared = "https://example.com/shared"
+
+        async def mock_arun_many(urls, config):
+            async def gen():
+                for url in urls:
+                    result = MagicMock()
+                    result.url = url
+                    result.success = True
+                    result.metadata = {}
+                    if url == start_url:
+                        links = [{"href": page_a}, {"href": page_b}]
+                    elif url in (page_a, page_b):
+                        links = [{"href": shared}]
+                    else:
+                        links = []
+                    result.links = {"internal": links, "external": []}
+                    yield result
+
+            return gen()
+
+        crawler = MagicMock()
+        crawler.arun_many = mock_arun_many
+        config = create_mock_config(stream=True)
+
+        scorer = CountingScorer()
+        strategy = BestFirstCrawlingStrategy(max_depth=2, url_scorer=scorer)
+
+        results = []
+        async for result in strategy._arun_best_first(start_url, crawler, config):
+            results.append(result)
+
+        assert scorer.call_counts.get(shared, 0) == 1
+        assert sum(1 for r in results if r.url == shared) == 1