fix: ensure dispatcher cleanup on crawler shutdown

bug_triage/repro_1278.py

@@ -0,0 +1,19 @@
+import asyncio
+from crawl4ai import AsyncWebCrawler
+
+async def main():
+    async with AsyncWebCrawler() as crawler:
+        # Target a page explicitly known for HTML tables
+        result = await crawler.arun(url="https://www.w3schools.com/html/html_tables.asp")
+
+        if result.success:
+            if "<table" not in result.cleaned_html.lower():
+                print("Bug reproduced: <table> tags are stripped from cleaned_html.")
+            else:
+                print("Fixed: Tables are preserved in cleaned_html.")
+        else:
+            print(f"Crawl failed: {result.error_message}")
+
+if __name__ == "__main__":
+    asyncio.run(main())
+

bug_triage/repro_1367.py

@@ -0,0 +1,25 @@
+import asyncio
+from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig
+
+async def main():
+    # Bug often triggers with persistent contexts and multiple concurrent navigations
+    browser_cfg = BrowserConfig(headless=True, use_persistent_context=True)
+    crawl_config = CrawlerRunConfig(wait_until="networkidle")
+
+    urls = ["https://example.com", "https://example.org", "https://example.net"]
+
+    async with AsyncWebCrawler(config=browser_cfg) as crawler:
+        print("Running async crawls to trigger ERR_ABORTED race condition...")
+        results = await crawler.arun_many(urls=urls, config=crawl_config)
+
+        reproduced = False
+        for res in results:
+            if not res.success and "ERR_ABORTED" in str(res.error_message):
+                print(f"Bug reproduced on {res.url}: {res.error_message}")
+                reproduced = True
+
+        if not reproduced:
+            print("Fixed or unable to reproduce: No ERR_ABORTED errors encountered.")
+
+if __name__ == "__main__":
+    asyncio.run(main())

bug_triage/repro_1455.py

@@ -0,0 +1,32 @@
+import os
+import asyncio
+from pydantic import BaseModel
+from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode, LLMConfig
+from crawl4ai import LLMExtractionStrategy
+
+class Product(BaseModel):
+    name: str
+
+async def main():
+    llm_strategy = LLMExtractionStrategy(
+        llm_config=LLMConfig(provider="openai/gpt-4o-mini", api_token=os.getenv('OPENAI_API_KEY', 'dummy')),
+        schema=Product.schema_json(),
+        extraction_type="schema",
+        instruction="Extract products."
+    )
+    crawl_config = CrawlerRunConfig(extraction_strategy=llm_strategy, cache_mode=CacheMode.ENABLED)
+
+    async with AsyncWebCrawler(config=BrowserConfig(headless=True)) as crawler:
+        print("First run (caching)...")
+        await crawler.arun(url="https://example.com", config=crawl_config)
+
+        print("Second run (cache hit)...")
+        result = await crawler.arun(url="https://example.com", config=crawl_config)
+
+        if not result.extracted_content:
+            print("Bug reproduced: extracted_content is empty on cache hit.")
+        else:
+            print("Fixed: extracted_content populated from cache.")
+
+if __name__ == "__main__":
+    asyncio.run(main())

bug_triage/repro_570.py

@@ -0,0 +1,19 @@
+import asyncio
+from crawl4ai import AsyncWebCrawler
+
+async def main():
+    async with AsyncWebCrawler() as crawler:
+        result = await crawler.arun(url="https://docs.crawl4ai.com/")
+
+        if result.success:
+            markdown = result.markdown or ""
+            if "<.>" in markdown or "<#>" in markdown:
+                print("Bug reproduced: Relative URLs are incorrectly formatted with brackets (e.g., <.>).")
+            else:
+                print("Fixed: No malformed relative URL brackets found in markdown.")
+        else:
+            print(f"Crawl failed: {result.error_message}")
+
+if __name__ == "__main__":
+    asyncio.run(main())
+

bug_triage/repro_699.py

@@ -0,0 +1,20 @@
+import asyncio
+from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
+
+async def main():
+    # Enable robots.txt checking
+    config = CrawlerRunConfig(check_robots_txt=True)
+
+    async with AsyncWebCrawler() as crawler:
+        # The path /awardsearch/advancedSearch.jsp is disallowed in nsf.gov/robots.txt
+        url = "https://www.nsf.gov/awardsearch/advancedSearch.jsp"
+        result = await crawler.arun(url=url, config=config)
+
+        if result.success:
+            print(f"Bug reproduced: Successfully crawled {url} despite robots.txt disallow.")
+        else:
+            print(f"Fixed/Expected behavior: Blocked from crawling. Message: {result.error_message}")
+
+if __name__ == "__main__":
+    asyncio.run(main())
+

crawl4ai/async_dispatcher.py

@@ -171,6 +171,34 @@ def __init__(
         self.memory_pressure_mode = False  # Flag to indicate when we're in memory pressure mode
         self.current_memory_percent = 0.0  # Track current memory usage
         self._high_memory_start_time: Optional[float] = None
+        self._stream_active_tasks = []
+        self._stream_memory_monitor = None
+        self._stream_cleanup_lock = asyncio.Lock()
+
+    async def cleanup(self) -> None:
+        async with self._stream_cleanup_lock:
+            for task in self._stream_active_tasks:
+                if not task.done():
+                    task.cancel()
+            if self._stream_active_tasks:
+                await asyncio.gather(*self._stream_active_tasks, return_exceptions=True)
+
+            while True:
+                try:
+                    self.task_queue.get_nowait()
+                except asyncio.QueueEmpty:
+                    break
+
+            if self._stream_memory_monitor is not None:
+                self._stream_memory_monitor.cancel()
+                await asyncio.gather(
+                    self._stream_memory_monitor, return_exceptions=True
+                )
+                self._stream_memory_monitor = None
+
+            self._stream_active_tasks = []
+            if self.monitor:
+                self.monitor.stop()
 
     async def _memory_monitor_task(self):
         """Background task to continuously monitor memory usage and update state"""
@@ -535,9 +563,11 @@ async def run_urls_stream(
     ) -> AsyncGenerator[CrawlerTaskResult, None]:
         self.crawler = crawler
         active_tasks = []
+        self._stream_active_tasks = active_tasks
 
         # Start the memory monitor task
         memory_monitor = asyncio.create_task(self._memory_monitor_task())
+        self._stream_memory_monitor = memory_monitor
 
         if self.monitor:
             self.monitor.start()
@@ -606,6 +636,7 @@ async def run_urls_stream(
 
                     # Update active tasks list
                     active_tasks = list(pending)
+                    self._stream_active_tasks = active_tasks
                 else:
                     # If no active tasks but still waiting, sleep briefly
                     await asyncio.sleep(self.check_interval / 2)
@@ -614,26 +645,7 @@ async def run_urls_stream(
                 await self._update_queue_priorities()
 
         finally:
-            # Cancel and await every task owned by this stream before returning
-            # control to the caller. Otherwise a closed stream can leave crawls
-            # using browser pages and contexts in the background.
-            for task in active_tasks:
-                if not task.done():
-                    task.cancel()
-            if active_tasks:
-                await asyncio.gather(*active_tasks, return_exceptions=True)
-
-            # Discard URLs that were queued by this stream but never started.
-            while True:
-                try:
-                    self.task_queue.get_nowait()
-                except asyncio.QueueEmpty:
-                    break
-
-            memory_monitor.cancel()
-            await asyncio.gather(memory_monitor, return_exceptions=True)
-            if self.monitor:
-                self.monitor.stop()
+            await self.cleanup()
 
 
 class SemaphoreDispatcher(BaseDispatcher):

crawl4ai/async_webcrawler.py

@@ -172,6 +172,7 @@ def __init__(
 
         self.url_seeder: Optional[AsyncUrlSeeder] = None
         self._domain_mapper: Optional[DomainMapper] = None
+        self._dispatchers = []
 
     async def start(self):
         """
@@ -194,7 +195,14 @@ async def close(self):
         1. Clean up browser resources
         2. Close any open pages and contexts
         """
-        await self.crawler_strategy.__aexit__(None, None, None)
+        try:
+            for dispatcher in self._dispatchers:
+                cleanup = getattr(dispatcher, "cleanup", None)
+                if cleanup:
+                    await cleanup()
+        finally:
+            self._dispatchers.clear()
+            await self.crawler_strategy.__aexit__(None, None, None)
 
     async def __aenter__(self):
         return await self.start()
@@ -1065,6 +1073,9 @@ async def _deep_crawl_stream():
                 ),
             )
 
+        if dispatcher not in self._dispatchers:
+            self._dispatchers.append(dispatcher)
+
         def transform_result(task_result):
             return (
                 setattr(
@@ -1107,14 +1118,23 @@ async def maybe_release_session():
 
         if stream:
             async def result_transformer():
+                inner_stream = dispatcher.run_urls_stream(
+                    crawler=self, urls=urls, config=config
+                )
                 try:
-                    async for task_result in dispatcher.run_urls_stream(
-                        crawler=self, urls=urls, config=config
-                    ):
+                    async for task_result in inner_stream:
                         yield transform_result(task_result)
+                except GeneratorExit:
+                    # Handle early stream breakage (e.g., break in a loop) explicitly
+                    pass
                 finally:
-                    # Auto-release session after streaming completes
-                    await maybe_release_session()
+                    try:
+                        cleanup = getattr(dispatcher, "cleanup", None)
+                        if cleanup:
+                            await cleanup()
+                    finally:
+                        # Auto-release session after streaming completes
+                        await maybe_release_session()
 
             return result_transformer()
         else:

crawl4ai/deep_crawling/bff_strategy.py

@@ -83,8 +83,6 @@ async def can_process_url(self, url: str, depth: int) -> bool:
                 raise ValueError("Missing scheme or netloc")
             if parsed.scheme not in ("http", "https"):
                 raise ValueError("Invalid scheme")
-            if "." not in parsed.netloc:
-                raise ValueError("Invalid domain")
         except Exception as e:
             self.logger.warning(f"Invalid URL: {url}, error: {e}")
             return False

crawl4ai/deep_crawling/bfs_strategy.py

@@ -70,8 +70,6 @@ async def can_process_url(self, url: str, depth: int) -> bool:
                 raise ValueError("Missing scheme or netloc")
             if parsed.scheme not in ("http", "https"):
                 raise ValueError("Invalid scheme")
-            if "." not in parsed.netloc:
-                raise ValueError("Invalid domain")
         except Exception as e:
             self.logger.warning(f"Invalid URL: {url}, error: {e}")
             return False