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,7 +20,7 @@
 from yarl import URL
 
 from crawlee._utils.urls import filter_url
-from crawlee._utils.web import is_status_code_successful
+from crawlee._utils.web import is_status_code_server_error, is_status_code_successful
 from crawlee.errors import ProxyError
 
 if TYPE_CHECKING:
@@ -46,6 +47,31 @@
 DEFAULT_MAX_DEPTH = 10
 """Default maximum depth of nested sitemaps to follow, guarding against malicious infinite sitemap chains."""
 
+SITEMAP_RETRY_DELAY = 1
+"""Seconds to wait before retrying a failed sitemap fetch."""
+
+
+def _is_successful_sitemap_status(status_code: int) -> bool:
+    """Return whether a sitemap response status is a successful 2xx."""
+    return HTTPStatus.OK <= status_code < HTTPStatus.MULTIPLE_CHOICES
+
+
+class _RetryableSitemapStatusError(Exception):
+    """Internal signal that a sitemap fetch returned a retryable HTTP status.
+
+    Raised inside the response context manager so the stream is closed before the retry delay.
+    """
+
+    def __init__(self, status_code: int) -> None:
+        super().__init__(f'HTTP status {status_code}')
+        self.status_code = status_code
+
+
+def _raise_for_retryable_sitemap_status(status_code: int) -> None:
+    """Raise `_RetryableSitemapStatusError` for HTTP statuses that warrant a fetch retry (408 and 5xx)."""
+    if status_code == HTTPStatus.REQUEST_TIMEOUT or is_status_code_server_error(status_code):
+        raise _RetryableSitemapStatusError(status_code)
+
 
 @dataclass()
 class SitemapUrl:
@@ -363,6 +389,7 @@ async def _fetch_and_process_sitemap(
     timeout: timedelta | None = None,
     emit_nested_sitemaps: bool,
     enqueue_strategy: EnqueueStrategy,
+    parsed_ok: list[bool],
 ) -> AsyncGenerator[SitemapUrl | NestedSitemap, None]:
     """Fetch a sitemap from a URL and process its content."""
     if 'url' not in source:
@@ -376,6 +403,13 @@ 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:
+                status_code = response.status_code
+                if not _is_successful_sitemap_status(status_code):
+                    # Retryable statuses raise and route to the retry handler below; the rest skip this source.
+                    _raise_for_retryable_sitemap_status(status_code)
+                    logger.warning(f'Skipping sitemap {sitemap_url} due to HTTP status code {status_code}.')
+                    return
+
                 # Determine content type and compression
                 content_type = response.headers.get('content-type', '')
 
@@ -456,15 +490,25 @@ async def _fetch_and_process_sitemap(
                                 yield result
                 finally:
                     parser.close()
-            break
+            parsed_ok.append(True)
 
+        except _RetryableSitemapStatusError as e:
+            if retries_left > 0:
+                logger.warning(
+                    f'Error fetching sitemap {sitemap_url}: HTTP status {e.status_code}. Retries left: {retries_left}'
+                )
+                await asyncio.sleep(SITEMAP_RETRY_DELAY)
+            else:
+                logger.warning(f'Failed to fetch sitemap {sitemap_url}, no retries left: HTTP status {e.status_code}.')
         except Exception as e:
             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
+                await asyncio.sleep(SITEMAP_RETRY_DELAY)
             else:
                 logger.exception(f'Failed to fetch sitemap {sitemap_url}, no retries left.')
                 raise
+        else:
+            return
 
 
 class Sitemap:
@@ -525,6 +569,10 @@ async def parse_sitemap(
 
     Default `ParseSitemapOptions.enqueue_strategy` is `same-hostname` which will skip cross-host URLs.
     Use strategy `all` to process all links.
+
+    URL sources answering with a non-2xx status are skipped: retryable statuses (408 and 5xx) are retried first,
+    others are skipped immediately. If at least one source fails with a fetch error after all retries and no source
+    succeeds, the first such error is raised once all sources have been processed.
     """
     # Set default options
     options = options or {}
@@ -537,6 +585,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,30 +609,43 @@ async def parse_sitemap(
                 enqueue_strategy=enqueue_strategy,
             ):
                 yield result
+            # Raw sources cannot fetch-fail, so they always count as successful for the all-sources-failed check.
+            successful_sources += 1
 
         elif source['type'] == 'url' and 'url' in source:
             # Add to visited set before processing to avoid duplicates
             if http_client is None:
                 raise RuntimeError('HttpClient must be provided for URL-based sitemap sources.')
 
             visited_sitemap_urls.add(source['url'])
+            parsed_ok: list[bool] = []
 
-            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,
+                    parsed_ok=parsed_ok,
+                ):
+                    yield result
+                if parsed_ok:
+                    successful_sources += 1
+            except Exception as e:
+                # Already logged by `_fetch_and_process_sitemap`; raised below only if no source succeeds.
+                source_errors.append(e)
         else:
             logger.warning(f'Invalid source configuration: {source}')
 
+    if source_errors and successful_sources == 0:
+        raise source_errors[0]
+
 
 async def _merge_async_generators(*generators: AsyncGenerator) -> AsyncGenerator:
     queue: asyncio.Queue = asyncio.Queue()

src/crawlee/request_loaders/_sitemap_request_loader.py

@@ -376,58 +376,64 @@ async def _load_sitemaps(self) -> None:
                 )
                 parsed_sitemap_url = URL(sitemap_url)
 
-                async with aclosing(
-                    parse_sitemap(
-                        [SitemapSource(type='url', url=sitemap_url)],
-                        self._http_client,
-                        proxy_info=self._proxy_info,
-                        options=parse_options,
-                    )
-                ) as sitemap_items:
-                    async for item in sitemap_items:
-                        if isinstance(item, NestedSitemap):
-                            # Add nested sitemap to queue
-                            if (
-                                item.loc not in state.pending_sitemap_urls
-                                and item.loc not in state.processed_sitemap_urls
-                            ):
-                                if current_depth >= DEFAULT_MAX_DEPTH:
-                                    logger.warning(
-                                        f'Skipping nested sitemap {item.loc!r}: max depth {DEFAULT_MAX_DEPTH} reached.'
-                                    )
-                                    continue
-                                if not self._passes_filters(item.loc, parsed_sitemap_url, 'nested sitemap'):
-                                    continue
-                                state.pending_sitemap_urls.append(item.loc)
-                                state.sitemap_depths[item.loc] = current_depth + 1
-                            continue
-
-                        if isinstance(item, SitemapUrl):
-                            url = item.loc
+                try:
+                    async with aclosing(
+                        parse_sitemap(
+                            [SitemapSource(type='url', url=sitemap_url)],
+                            self._http_client,
+                            proxy_info=self._proxy_info,
+                            options=parse_options,
+                        )
+                    ) as sitemap_items:
+                        async for item in sitemap_items:
+                            if isinstance(item, NestedSitemap):
+                                # Add nested sitemap to queue
+                                if (
+                                    item.loc not in state.pending_sitemap_urls
+                                    and item.loc not in state.processed_sitemap_urls
+                                ):
+                                    if current_depth >= DEFAULT_MAX_DEPTH:
+                                        logger.warning(
+                                            f'Skipping nested sitemap {item.loc!r}: '
+                                            f'max depth {DEFAULT_MAX_DEPTH} reached.'
+                                        )
+                                        continue
+                                    if not self._passes_filters(item.loc, parsed_sitemap_url, 'nested sitemap'):
+                                        continue
+                                    state.pending_sitemap_urls.append(item.loc)
+                                    state.sitemap_depths[item.loc] = current_depth + 1
+                                continue
 
-                            state = await self._get_state()
+                            if isinstance(item, SitemapUrl):
+                                url = item.loc
 
-                            # Skip if already processed
-                            if url in state.current_sitemap_processed_urls:
-                                continue
+                                state = await self._get_state()
 
-                            # Check if URL should be included
-                            if not self._check_url_patterns(url, self._include, self._exclude):
-                                continue
+                                # Skip if already processed
+                                if url in state.current_sitemap_processed_urls:
+                                    continue
 
-                            if not self._passes_filters(url, parsed_sitemap_url, 'sitemap URL'):
-                                continue
+                                # Check if URL should be included
+                                if not self._check_url_patterns(url, self._include, self._exclude):
+                                    continue
 
-                            # Check if we have capacity in the queue
-                            await self._queue_has_capacity.wait()
+                                if not self._passes_filters(url, parsed_sitemap_url, 'sitemap URL'):
+                                    continue
 
-                            async with self._queue_lock:
-                                state.url_queue.append(url)
-                                state.current_sitemap_processed_urls.add(url)
-                                state.total_count += 1
-                                if len(state.url_queue) >= self._max_buffer_size:
-                                    # Notify that the queue is full
-                                    self._queue_has_capacity.clear()
+                                # Check if we have capacity in the queue
+                                await self._queue_has_capacity.wait()
+
+                                async with self._queue_lock:
+                                    state.url_queue.append(url)
+                                    state.current_sitemap_processed_urls.add(url)
+                                    state.total_count += 1
+                                    if len(state.url_queue) >= self._max_buffer_size:
+                                        # Notify that the queue is full
+                                        self._queue_has_capacity.clear()
+                except Exception:
+                    # A sitemap that fails to load is marked processed and skipped, so one dead sitemap
+                    # does not abandon the remaining pending sitemaps.
+                    logger.warning(f'Failed to load sitemap {sitemap_url}, skipping it.')
 
                 # Clear current sitemap after processing
                 state = await self._get_state()
@@ -442,5 +448,6 @@ async def _load_sitemaps(self) -> None:
             state.completed = True
 
         except Exception:
+            # Not re-raised: nothing retrieves the task's exception (`is_finished` only checks doneness), so it
+            # would only resurface as an "exception was never retrieved" asyncio error when the task is collected.
             logger.exception('Error loading sitemaps')
-            raise