fix: complete SessionError lifecycle and honor reclaim forefront

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
                     )

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)

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

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)