diff --git a/src/crawlee/crawlers/_adaptive_playwright/_adaptive_playwright_crawler.py b/src/crawlee/crawlers/_adaptive_playwright/_adaptive_playwright_crawler.py index 2abed1399e..350ec93d15 100644 --- a/src/crawlee/crawlers/_adaptive_playwright/_adaptive_playwright_crawler.py +++ b/src/crawlee/crawlers/_adaptive_playwright/_adaptive_playwright_crawler.py @@ -30,6 +30,7 @@ from crawlee.crawlers._beautifulsoup._beautifulsoup_parser import BeautifulSoupParser from crawlee.crawlers._parsel._parsel_parser import ParselParser from crawlee.crawlers._playwright._playwright_crawler import _PlaywrightCrawlerAdditionalOptions +from crawlee.errors import SessionError from crawlee.statistics import Statistics, StatisticsState from ._adaptive_playwright_crawler_statistics import AdaptivePlaywrightCrawlerStatisticState @@ -396,6 +397,10 @@ async def _run_request_handler(self, context: BasicCrawlingContext) -> None: self._context_result_map[context] = static_run.result return if static_run.exception: + # SessionError means the session is blocked. Falling through to the browser with the same + # session would keep using a bad session; re-raise so BasicCrawler can rotate/retire it. + if isinstance(static_run.exception, SessionError): + raise static_run.exception context.log.error( msg=f'Static crawler: failed for {context.request.url}', exc_info=static_run.exception ) diff --git a/src/crawlee/crawlers/_basic/_basic_crawler.py b/src/crawlee/crawlers/_basic/_basic_crawler.py index e6944ee2b0..0e2c22dadd 100644 --- a/src/crawlee/crawlers/_basic/_basic_crawler.py +++ b/src/crawlee/crawlers/_basic/_basic_crawler.py @@ -1180,7 +1180,7 @@ async def _handle_request_retries( await self._mark_request_as_handled(request) return - await request_manager.reclaim_request(request) + await request_manager.reclaim_request(request, forefront=request.forefront) else: request.state = RequestState.ERROR await self._mark_request_as_handled(request) @@ -1474,22 +1474,33 @@ async def __run_task_function(self) -> None: if not session: raise RuntimeError('SessionError raised in a crawling context without a session') from session_error + new_request = None if self._error_handler: - await self._error_handler(context, session_error) + try: + new_request = await self._error_handler(context, session_error) + except Exception as e: + raise UserDefinedErrorHandlerError('Exception thrown in user-defined request error handler') from e + + if new_request is not None and new_request != request: + await request_manager.add_request(new_request) + await self._mark_request_as_handled(request) + session.retire() + return if self._should_retry_request(context, session_error): exc_only = ''.join(traceback.format_exception_only(session_error)).strip() self._logger.warning('Encountered "%s", rotating session and retrying...', exc_only) - if session: - session.retire() + session.retire() # Increment session rotation count. request.session_rotation_count = (request.session_rotation_count or 0) + 1 - await request_manager.reclaim_request(request) + await request_manager.reclaim_request(request, forefront=request.forefront) await self._statistics.error_tracker_retry.add(error=session_error, context=context) else: + # Exhausted rotations: retire the blocked session so it is not reused from the pool. + session.retire() await self._mark_request_as_handled(request) await self._handle_failed_request(context, session_error) diff --git a/tests/unit/crawlers/_adaptive_playwright/test_adaptive_playwright_crawler.py b/tests/unit/crawlers/_adaptive_playwright/test_adaptive_playwright_crawler.py index dea77ac1d4..cb5452541f 100644 --- a/tests/unit/crawlers/_adaptive_playwright/test_adaptive_playwright_crawler.py +++ b/tests/unit/crawlers/_adaptive_playwright/test_adaptive_playwright_crawler.py @@ -6,7 +6,7 @@ from dataclasses import dataclass from datetime import timedelta from itertools import cycle -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING, Any, cast from unittest.mock import Mock, call, patch import pytest @@ -25,10 +25,12 @@ RenderingTypePrediction, RenderingTypePredictor, ) +from crawlee.crawlers._adaptive_playwright._adaptive_playwright_crawler import SubCrawlerRun from crawlee.crawlers._adaptive_playwright._adaptive_playwright_crawler_statistics import ( AdaptivePlaywrightCrawlerStatisticState, ) from crawlee.crawlers._adaptive_playwright._adaptive_playwright_crawling_context import AdaptiveContextError +from crawlee.errors import SessionError from crawlee.sessions import SessionPool from crawlee.statistics import Statistics from crawlee.storage_clients import SqlStorageClient @@ -949,3 +951,44 @@ def test_import_error_handled(optional_module_name: str) -> None: with pytest.raises(ImportError): from crawlee.crawlers import AdaptivePlaywrightCrawler # noqa: F401 PLC0415 + + +async def test_static_session_error_propagates_for_rotation(server_url: URL) -> None: + """SessionError from the static path must propagate so BasicCrawler can rotate the session. + + Other static exceptions still fall through to the browser; SessionError specifically means the + session is blocked and must not be reused for a browser fallback. + """ + predictor = _SimpleRenderingTypePredictor(detection_probability_recommendation=cycle([0])) + crawler = AdaptivePlaywrightCrawler.with_beautifulsoup_static_parser( + rendering_type_predictor=predictor, + max_session_rotations=3, + ) + + session_ids: list[str] = [] + original_crawl_one = crawler._crawl_one + + async def crawl_one_raising_session_error( + *, + rendering_type: RenderingType, + context: Any, + **kwargs: Any, + ) -> Any: + if rendering_type == 'static': + if context.session: + session_ids.append(context.session.id) + return SubCrawlerRun(exception=SessionError('static blocked')) + return await original_crawl_one(rendering_type=rendering_type, context=context, **kwargs) + + crawler._crawl_one = crawl_one_raising_session_error # type: ignore[method-assign] + + @crawler.router.default_handler + async def handler(context: AdaptivePlaywrightCrawlingContext) -> None: + pass + + stats = await crawler.run([str(server_url)]) + + assert len(session_ids) == 3 + assert len(set(session_ids)) == 3 + assert crawler.statistics.state.browser_request_handler_runs == 0 + assert stats.requests_failed == 1 diff --git a/tests/unit/crawlers/_basic/test_basic_crawler.py b/tests/unit/crawlers/_basic/test_basic_crawler.py index 763969b3f9..8cc62da725 100644 --- a/tests/unit/crawlers/_basic/test_basic_crawler.py +++ b/tests/unit/crawlers/_basic/test_basic_crawler.py @@ -274,6 +274,89 @@ async def error_handler(context: BasicCrawlingContext, error: Exception) -> None assert error_handler_mock.call_count == 1 +async def test_session_error_handler_can_replace_request() -> None: + """`error_handler` return value must be honored for SessionError, same as for regular errors.""" + queue = await RequestQueue.open() + crawler = BasicCrawler(request_manager=queue, max_session_rotations=3) + + request = Request.from_url('https://a.placeholder.com') + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + if '|recovered' in context.request.unique_key: + return + raise SessionError('blocked') + + @crawler.error_handler + async def error_handler(context: BasicCrawlingContext, error: Exception) -> Request | None: + assert isinstance(error, SessionError) + return Request.from_url( + context.request.url, + unique_key=f'{context.request.unique_key}|recovered', + ) + + await crawler.run([request]) + + original_request = await queue.get_request(request.unique_key) + recovered_request = await queue.get_request(f'{request.unique_key}|recovered') + + assert original_request is not None + assert original_request.was_already_handled + assert recovered_request is not None + assert recovered_request.state == RequestState.DONE + assert recovered_request.was_already_handled + + await queue.drop() + + +async def test_session_retired_when_rotations_exhausted() -> None: + """When session rotations are exhausted, the blocked session must be retired.""" + retired_sessions: list[Session] = [] + + crawler = BasicCrawler(max_session_rotations=2) + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + assert context.session is not None + raise SessionError('blocked') + + @crawler.failed_request_handler + async def failed_request_handler(context: BasicCrawlingContext, error: Exception) -> None: + assert context.session is not None + retired_sessions.append(context.session) + + await crawler.run(['https://a.placeholder.com']) + + assert len(retired_sessions) == 1 + assert not retired_sessions[0].is_usable + + +async def test_reclaim_uses_request_forefront_flag() -> None: + """Retries must reclaim with `request.forefront` so tiered-proxy priority retries stay at the front.""" + queue = await RequestQueue.open() + reclaim_calls: list[bool] = [] + original_reclaim = queue.reclaim_request + + async def tracking_reclaim(request: Request, *, forefront: bool = False) -> Any: + reclaim_calls.append(forefront) + return await original_reclaim(request, forefront=forefront) + + queue.reclaim_request = tracking_reclaim # type: ignore[method-assign] + + crawler = BasicCrawler(request_manager=queue, max_request_retries=1) + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + context.request.forefront = True + raise RuntimeError('retry me') + + await crawler.run([Request.from_url('https://a.placeholder.com')]) + + assert reclaim_calls == [True] + + await queue.drop() + + async def test_handles_error_in_error_handler() -> None: crawler = BasicCrawler(max_request_retries=3)