fix: validate HTTP status when fetching sitemaps

src/crawlee/_utils/sitemap.py

@@ -9,6 +9,7 @@
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from hashlib import sha256
+from http import HTTPStatus
 from logging import getLogger
 from typing import TYPE_CHECKING, Literal, TypedDict
 from xml.sax import SAXParseException
@@ -19,8 +20,8 @@
 from yarl import URL
 
 from crawlee._utils.urls import filter_url
-from crawlee._utils.web import is_status_code_successful
-from crawlee.errors import ProxyError
+from crawlee._utils.web import is_status_code_server_error, is_status_code_successful
+from crawlee.errors import HttpStatusCodeError, ProxyError
 
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator
@@ -47,6 +48,21 @@
 """Default maximum depth of nested sitemaps to follow, guarding against malicious infinite sitemap chains."""
 
 
+def _raise_for_sitemap_status(status_code: int) -> None:
+    """Raise `HttpStatusCodeError` if the sitemap response status is not 2xx."""
+    if not HTTPStatus.OK <= status_code < HTTPStatus.MULTIPLE_CHOICES:
+        raise HttpStatusCodeError('Error status code returned while fetching sitemap', status_code)
+
+
+def _is_retryable_sitemap_status(status_code: int) -> bool:
+    """Return whether a sitemap response status should be retried."""
+    return (
+        HTTPStatus.MULTIPLE_CHOICES <= status_code < HTTPStatus.BAD_REQUEST
+        or status_code in (HTTPStatus.REQUEST_TIMEOUT, HTTPStatus.TOO_MANY_REQUESTS)
+        or is_status_code_server_error(status_code)
+    )
+
+
 @dataclass()
 class SitemapUrl:
     loc: str
@@ -376,6 +392,8 @@ async def _fetch_and_process_sitemap(
             async with http_client.stream(
                 sitemap_url, method='GET', headers=SITEMAP_HEADERS, proxy_info=proxy_info, timeout=timeout
             ) as response:
+                _raise_for_sitemap_status(response.status_code)
+
                 # Determine content type and compression
                 content_type = response.headers.get('content-type', '')
 
@@ -459,9 +477,14 @@ async def _fetch_and_process_sitemap(
             break
 
         except Exception as e:
+            if isinstance(e, HttpStatusCodeError) and not _is_retryable_sitemap_status(e.status_code):
+                logger.warning(f'Skipping sitemap {sitemap_url} due to HTTP status code {e.status_code}.')
+                break
             if retries_left > 0:
                 logger.warning(f'Error fetching sitemap {sitemap_url}: {e}. Retries left: {retries_left}')
                 await asyncio.sleep(1)  # Brief pause before retry
+            elif isinstance(e, HttpStatusCodeError):
+                logger.warning(f'Failed to fetch sitemap {sitemap_url}, no retries left: {e}')
             else:
                 logger.exception(f'Failed to fetch sitemap {sitemap_url}, no retries left.')
                 raise
@@ -537,6 +560,8 @@ async def parse_sitemap(
     # Setup working state
     sources = list(initial_sources)
     visited_sitemap_urls: set[str] = set()
+    successful_sources = 0
+    source_errors: list[Exception] = []
 
     # Process sources until the queue is empty
     while sources:
@@ -559,6 +584,7 @@ async def parse_sitemap(
                 enqueue_strategy=enqueue_strategy,
             ):
                 yield result
+            successful_sources += 1
 
         elif source['type'] == 'url' and 'url' in source:
             # Add to visited set before processing to avoid duplicates
@@ -567,22 +593,30 @@ async def parse_sitemap(
 
             visited_sitemap_urls.add(source['url'])
 
-            async for result in _fetch_and_process_sitemap(
-                http_client=http_client,
-                source=source,
-                depth=depth,
-                visited_sitemap_urls=visited_sitemap_urls,
-                sources=sources,
-                retries_left=sitemap_retries,
-                emit_nested_sitemaps=emit_nested_sitemaps,
-                enqueue_strategy=enqueue_strategy,
-                proxy_info=proxy_info,
-                timeout=timeout,
-            ):
-                yield result
+            try:
+                async for result in _fetch_and_process_sitemap(
+                    http_client=http_client,
+                    source=source,
+                    depth=depth,
+                    visited_sitemap_urls=visited_sitemap_urls,
+                    sources=sources,
+                    retries_left=sitemap_retries,
+                    emit_nested_sitemaps=emit_nested_sitemaps,
+                    enqueue_strategy=enqueue_strategy,
+                    proxy_info=proxy_info,
+                    timeout=timeout,
+                ):
+                    yield result
+                successful_sources += 1
+            except Exception as e:
+                source_errors.append(e)
+                logger.warning(f'Failed to process sitemap source {source["url"]}: {e}')
         else:
             logger.warning(f'Invalid source configuration: {source}')
 
+    if source_errors and successful_sources == 0:
+        raise source_errors[-1]
+
 
 async def _merge_async_generators(*generators: AsyncGenerator) -> AsyncGenerator:
     queue: asyncio.Queue = asyncio.Queue()

src/crawlee/request_loaders/_request_loader.py

@@ -39,14 +39,14 @@ async def is_empty(self) -> bool:
 
     @abstractmethod
     async def is_finished(self) -> bool:
-        """Return True if all requests have been handled."""
+        """Return True if all requests have been handled, or raise if loading failed after pending requests drain."""
 
     @abstractmethod
     async def fetch_next_request(self) -> Request | None:
         """Return the next request to be processed, or `None` if there are no more pending requests.
 
         The method should return `None` if and only if `is_finished` would return `True`. In other cases, the method
-        should wait until a request appears.
+        should wait until a request appears. It can raise a loading error after all pending requests have been handled.
         """
 
     @abstractmethod

src/crawlee/request_loaders/_sitemap_request_loader.py

@@ -214,7 +214,11 @@ async def is_empty(self) -> bool:
     async def is_finished(self) -> bool:
         """Check if all URLs have been processed."""
         state = await self._get_state()
-        return not state.url_queue and len(state.in_progress) == 0 and self._loading_task.done()
+        if state.url_queue or state.in_progress:
+            return False
+        if self._loading_task.done() and not self._loading_task.cancelled():
+            self._loading_task.result()
+        return self._loading_task.done()
 
     @override
     async def fetch_next_request(self) -> Request | None:

tests/unit/_utils/test_sitemap.py

@@ -14,6 +14,7 @@
     DEFAULT_MAX_DEPTH,
     ParseSitemapOptions,
     Sitemap,
+    SitemapSource,
     SitemapUrl,
     _TxtSitemapParser,
     _XMLSaxSitemapHandler,
@@ -22,7 +23,13 @@
     parse_sitemap,
 )
 from crawlee.http_clients._base import HttpClient, HttpResponse
-from tests.unit.utils import DEFAULT_URL, get_basic_results, get_basic_sitemap
+from tests.unit.utils import (
+    DEFAULT_URL,
+    get_basic_results,
+    get_basic_sitemap,
+    make_status_stream_client,
+    sleep_without_delay,
+)
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator, Callable
@@ -60,6 +67,7 @@ async def read_stream() -> 'AsyncIterator[bytes]':
             yield body
 
         response = MagicMock(spec=HttpResponse)
+        response.status_code = 200
         response.headers = {'content-type': 'application/xml; charset=utf-8'}
         response.read_stream = read_stream
         yield cast('HttpResponse', response)
@@ -81,6 +89,7 @@ async def read_stream() -> 'AsyncIterator[bytes]':
             yield body_for_url(url)
 
         response = MagicMock(spec=HttpResponse)
+        response.status_code = 200
         response.headers = {'content-type': 'application/xml; charset=utf-8'}
         response.read_stream = read_stream
         yield cast('HttpResponse', response)
@@ -337,8 +346,9 @@ async def test_malformed_sitemap_keeps_urls() -> None:
     assert sitemap.urls == [f'{DEFAULT_URL}first', f'{DEFAULT_URL}second']
 
 
-async def test_sitemap_fetch_retries_on_transient_error() -> None:
+async def test_sitemap_fetch_retries_on_transient_error(monkeypatch: pytest.MonkeyPatch) -> None:
     """Transient fetch errors are retried up to `sitemap_retries` times before giving up."""
+    monkeypatch.setattr('crawlee._utils.sitemap.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay))
     client, attempts = _make_flaky_stream_client(get_basic_sitemap().encode(), fail_times=2)
 
     items = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)]
@@ -347,8 +357,9 @@ async def test_sitemap_fetch_retries_on_transient_error() -> None:
     assert {item.loc for item in items} == get_basic_results()
 
 
-async def test_sitemap_fetch_raises_after_retries_exhausted() -> None:
+async def test_sitemap_fetch_raises_after_retries_exhausted(monkeypatch: pytest.MonkeyPatch) -> None:
     """A persistent fetch error is raised to the caller once all retries are exhausted."""
+    monkeypatch.setattr('crawlee._utils.sitemap.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay))
     client, attempts = _make_flaky_stream_client(get_basic_sitemap().encode(), fail_times=10)
 
     with pytest.raises(ConnectionError):
@@ -357,6 +368,78 @@ async def test_sitemap_fetch_raises_after_retries_exhausted() -> None:
     assert len(attempts) == 3
 
 
+async def test_sitemap_fetch_retries_retryable_http_status(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Retryable HTTP errors are retried before parsing a successful response."""
+    monkeypatch.setattr('crawlee._utils.sitemap.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay))
+    client, attempts = make_status_stream_client([(503, b''), (503, b''), (200, get_basic_sitemap().encode())])
+
+    items = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)]
+
+    assert attempts == [503, 503, 200]
+    assert {item.loc for item in items} == get_basic_results()
+
+
+async def test_sitemap_fetch_skips_http_error_after_retries_exhausted(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A persistent retryable HTTP error is skipped once retries are exhausted."""
+    monkeypatch.setattr('crawlee._utils.sitemap.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay))
+    client, attempts = make_status_stream_client([(503, b'')])
+
+    items = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)]
+
+    assert attempts == [503, 503, 503]
+    assert items == []
+
+
+async def test_sitemap_fetch_does_not_retry_terminal_http_status() -> None:
+    """Terminal HTTP errors are skipped without parsing their response body or retrying."""
+    client, attempts = make_status_stream_client([(404, get_basic_sitemap().encode())])
+
+    items = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)]
+
+    assert attempts == [404]
+    assert items == []
+
+
+async def test_sitemap_fetch_retries_redirect_then_skips(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Redirect responses that reach the parser are retried and skipped after exhaustion."""
+    monkeypatch.setattr('crawlee._utils.sitemap.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay))
+    client, attempts = make_status_stream_client([(302, get_basic_sitemap().encode())])
+
+    items = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)]
+
+    assert attempts == [302, 302, 302]
+    assert items == []
+
+
+async def test_sitemap_partial_http_failure_keeps_healthy_source() -> None:
+    """An HTTP failure in one source does not discard URLs from another source."""
+    client, attempts = make_status_stream_client([(200, get_basic_sitemap().encode()), (404, b'')])
+    sources: list[SitemapSource] = [
+        {'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'},
+        {'type': 'url', 'url': f'{DEFAULT_URL}missing.xml'},
+    ]
+
+    items = [item async for item in parse_sitemap(sources, client)]
+
+    assert attempts == [200, 404]
+    assert {item.loc for item in items} == get_basic_results()
+
+
+async def test_sitemap_partial_fetch_failure_keeps_healthy_source(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A fetch exception in one source is suppressed when another source succeeds."""
+    monkeypatch.setattr('crawlee._utils.sitemap.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay))
+    client, attempts = _make_flaky_stream_client(get_basic_sitemap().encode(), fail_times=3)
+    sources: list[SitemapSource] = [
+        {'type': 'url', 'url': f'{DEFAULT_URL}broken.xml'},
+        {'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'},
+    ]
+
+    items = [item async for item in parse_sitemap(sources, client)]
+
+    assert len(attempts) == 4
+    assert {item.loc for item in items} == get_basic_results()
+
+
 async def test_gzip_bomb_sitemap_truncated_at_size_cap(monkeypatch: pytest.MonkeyPatch) -> None:
     """A gzip sitemap inflating past the size cap is truncated instead of being decompressed without bound."""
     monkeypatch.setattr('crawlee._utils.sitemap.MAX_SITEMAP_SIZE', 64 * 1024)
@@ -388,6 +471,7 @@ async def read_stream() -> 'AsyncIterator[bytes]':
                 yield b'\x00' * 65536
 
         response = MagicMock(spec=HttpResponse)
+        response.status_code = 200
         response.headers = {'content-type': 'application/gzip'}
         response.read_stream = read_stream
         yield cast('HttpResponse', response)
@@ -413,6 +497,7 @@ async def read_stream() -> 'AsyncIterator[bytes]':
             yield body
 
         response = MagicMock(spec=HttpResponse)
+        response.status_code = 200
         response.headers = {'content-type': 'application/gzip'}
         response.read_stream = read_stream
         yield cast('HttpResponse', response)