fix(scrapy): stop silently dropping in-flight requests and redirects

docs/03_guides/06_scrapy.mdx

@@ -104,7 +104,7 @@ The following example shows a Scrapy Actor that scrapes page titles and enqueues
 
 ## Dealing with imminent migration to another host
 
-Under some circumstances, the platform may decide to [migrate your Actor](https://docs.apify.com/academy/expert-scraping-with-apify/migrations-maintaining-state) from one piece of infrastructure to another while it's in progress. While [Crawlee](https://crawlee.dev/python)-based projects can pause and resume their work after a restart, achieving the same with a Scrapy-based project can be challenging.
+Under some circumstances, the platform may decide to [migrate your Actor](https://docs.apify.com/academy/expert-scraping-with-apify/migrations-maintaining-state) from one piece of infrastructure to another while the run is in progress. Requests that Scrapy hasn't finished when the run stops stay unhandled in the request queue, so the next run picks them up and downloads them from scratch. A Scrapy-based project doesn't resume where it left off the way a [Crawlee](https://crawlee.dev/python)-based one does, so items that their callbacks already pushed can land in the dataset twice.
 
 As a workaround for this issue (tracked as [apify/actor-templates#303](https://github.com/apify/actor-templates/issues/303)), turn on caching with `HTTPCACHE_ENABLED` and set `HTTPCACHE_EXPIRATION_SECS` to at least a few minutes—the exact value depends on your use case. If your Actor gets migrated and restarted, the subsequent run will hit the cache, making it fast and avoiding unnecessary resource consumption.
 

src/apify/scrapy/_async_thread.py

@@ -12,6 +12,9 @@
 
 logger = getLogger(__name__)
 
+SUBMITTED_PRUNE_THRESHOLD = 128
+"""How many `submit_coro` futures may pile up before the finished ones are dropped from the tracking list."""
+
 
 class AsyncThread:
     """Run an asyncio event loop in a dedicated background thread.
@@ -26,6 +29,9 @@ def __init__(self, default_timeout: timedelta = timedelta(seconds=60)) -> None:
         self._default_timeout = default_timeout
         self._eventloop = asyncio.new_event_loop()
 
+        self._submitted: list[futures.Future] = []
+        """Futures of the coroutines submitted via `submit_coro` that may still be running."""
+
         # Start the event loop in a dedicated daemon thread.
         self._thread = threading.Thread(
             target=self._start_event_loop,
@@ -74,6 +80,55 @@ def run_coro(
             future.cancel()
             raise
 
+    def submit_coro(self, coro: Coroutine) -> futures.Future:
+        """Schedule a coroutine on the event loop without waiting for its result.
+
+        Use this for work nothing depends on, so the calling thread is not blocked by the round trip. Failures
+        are logged, as there is no caller to propagate them to, and `close` cancels whatever is still pending -
+        call `wait_for_submitted` first if that matters.
+
+        Args:
+            coro: The coroutine to run.
+
+        Returns:
+            The future of the scheduled coroutine, for callers that want to inspect its outcome later.
+
+        Raises:
+            RuntimeError: If the event loop has been closed.
+        """
+        if self._eventloop.is_closed():
+            raise RuntimeError(f'The coroutine {coro} cannot be executed because the event loop is closed.')
+
+        # Callers may go a long time between `wait_for_submitted` calls, so without pruning the list would hold
+        # every coroutine ever submitted, with its result.
+        if len(self._submitted) >= SUBMITTED_PRUNE_THRESHOLD:
+            self._submitted = [submitted for submitted in self._submitted if not submitted.done()]
+
+        future = asyncio.run_coroutine_threadsafe(coro, self._eventloop)
+        future.add_done_callback(self._log_failure)
+        self._submitted.append(future)
+
+        return future
+
+    def wait_for_submitted(self, timeout: timedelta | None = None) -> None:
+        """Block until the coroutines submitted via `submit_coro` have finished.
+
+        Use this before anything that would observe their effects, or before `close`, which cancels whatever is
+        still running. Coroutines that do not finish within the timeout stay tracked for the next call.
+
+        Args:
+            timeout: The maximum time to wait for the submitted coroutines. Pass `None` to use the
+                `default_timeout` passed to the constructor.
+        """
+        if timeout is None:
+            timeout = self._default_timeout
+
+        self._submitted = list(futures.wait(self._submitted, timeout=timeout.total_seconds()).not_done)
+
+        # Callers rely on the effects having landed, so a timeout has to be visible.
+        if self._submitted:
+            logger.warning(f'{len(self._submitted)} submitted coroutines did not finish within the timeout.')
+
     def close(self, timeout: timedelta | None = None) -> None:
         """Close the event loop and its thread gracefully.
 
@@ -110,6 +165,15 @@ def close(self, timeout: timedelta | None = None) -> None:
                 logger.warning('Event loop thread did not exit cleanly! Forcing shutdown...')
                 self._force_exit_event_loop()
 
+    @staticmethod
+    def _log_failure(future: futures.Future) -> None:
+        """Log the failure of a coroutine submitted via `submit_coro`."""
+        if future.cancelled():
+            return
+
+        if (exc := future.exception()) is not None:
+            logger.error('A coroutine submitted to the event loop failed.', exc_info=exc)
+
     def _start_event_loop(self) -> None:
         """Set up and run the asyncio event loop in the dedicated thread."""
         asyncio.set_event_loop(self._eventloop)

src/apify/scrapy/requests.py

@@ -2,7 +2,7 @@
 
 from copy import deepcopy
 from logging import getLogger
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, get_args
 
 from scrapy import Request as ScrapyRequest
 from scrapy import Spider
@@ -11,11 +11,15 @@
 from scrapy.utils.request import request_from_dict
 
 from crawlee._request import UserData
-from crawlee._types import HttpHeaders
+from crawlee._types import HttpHeaders, HttpMethod
+from crawlee._utils.requests import compute_unique_key
 
 from ._serialization import decode_from_json, encode_to_json
 from apify import Request as ApifyRequest
 
+if TYPE_CHECKING:
+    from typing_extensions import TypeIs
+
 logger = getLogger(__name__)
 
 
@@ -47,6 +51,38 @@ def _ensure_known_request_class(request_dict: dict[str, Any]) -> None:
         )
 
 
+def _is_http_method(method: str) -> TypeIs[HttpMethod]:
+    """Tell whether `method` is one of the HTTP methods the RQ accepts; Scrapy itself takes any string."""
+    return method in get_args(HttpMethod)
+
+
+def _compute_fingerprint(scrapy_request: ScrapyRequest) -> str:
+    """Identify the request an RQ unique key was minted for.
+
+    `to_scrapy_request` stamps this beside the unique key so `to_apify_request` can tell a request that came out
+    of the RQ from one Scrapy derived from it. It covers the URL, the method and the body, so a redirect, a
+    method switch and a changed body all read as a different request rather than the parent.
+
+    Headers are left out even though the unique key covers them: `DefaultHeadersMiddleware` and
+    `UserAgentMiddleware` call `headers.setdefault()` on the request in place before the scheduler sees it
+    again, so an otherwise untouched request would no longer match its own stamp.
+    """
+    method = scrapy_request.method
+    if not _is_http_method(method):
+        raise ValueError(
+            f'Unsupported HTTP method {method!r} in {scrapy_request}; '
+            f'the request queue accepts {", ".join(get_args(HttpMethod))}.'
+        )
+
+    return compute_unique_key(
+        url=scrapy_request.url,
+        method=method,
+        payload=scrapy_request.body,
+        keep_url_fragment=False,
+        use_extended_unique_key=True,
+    )
+
+
 def to_apify_request(scrapy_request: ScrapyRequest, spider: Spider) -> ApifyRequest | None:
     """Convert a Scrapy request to an Apify request.
 
@@ -76,8 +112,14 @@ def to_apify_request(scrapy_request: ScrapyRequest, spider: Spider) -> ApifyRequ
     try:
         if scrapy_request.dont_filter:
             request_kwargs['always_enqueue'] = True
-        elif scrapy_request.meta.get('apify_request_unique_key'):
-            request_kwargs['unique_key'] = scrapy_request.meta['apify_request_unique_key']
+        elif unique_key := scrapy_request.meta.get('apify_request_unique_key'):
+            # Reuse the RQ unique key only while this is still the request it was minted for. Redirects
+            # (`Request.replace()`) and spiders forwarding `meta` to another URL inherit the stamp, and reusing
+            # it there deduplicates the derived request against its parent. A stamp without a fingerprint was
+            # set by hand, so it is taken at face value.
+            fingerprint = _compute_fingerprint(scrapy_request)
+            if scrapy_request.meta.get('apify_request_fingerprint', fingerprint) == fingerprint:
+                request_kwargs['unique_key'] = unique_key
 
         # Serialize the Scrapy request now, before `Request.from_url()` runs below. `from_url()` mutates the
         # `user_data` dict it receives in place (it injects a live `CrawleeRequestData` under `__crawlee`), and that
@@ -113,8 +155,9 @@ def to_apify_request(scrapy_request: ScrapyRequest, spider: Spider) -> ApifyRequ
         # the old behavior, which dropped such requests entirely.
         if isinstance(scrapy_request.headers, Headers):
             try:
-                headers = cast('dict[str, str]', dict(scrapy_request.headers.to_unicode_dict()))
-                request_kwargs['headers'] = HttpHeaders(headers)
+                # `to_unicode_dict()` yields str keys and values, but Scrapy types it as `UserDict[str | bytes, Any]`,
+                # so the mapping goes through the model's validator instead of a cast.
+                request_kwargs['headers'] = HttpHeaders.model_validate(scrapy_request.headers.to_unicode_dict())
             except UnicodeDecodeError:
                 logger.warning(
                     'Could not represent Scrapy request headers as Apify request headers (non-UTF-8 values); '
@@ -165,7 +208,7 @@ def to_scrapy_request(apify_request: ApifyRequest, spider: Spider) -> ScrapyRequ
     Returns:
         The converted Scrapy request.
     """
-    if not isinstance(cast('Any', apify_request), ApifyRequest):
+    if not isinstance(apify_request, ApifyRequest):
         raise TypeError('apify_request must be an apify.Request instance')
 
     # If the apify_request comes from the Scrapy
@@ -187,21 +230,14 @@ def to_scrapy_request(apify_request: ApifyRequest, spider: Spider) -> ScrapyRequ
         if not isinstance(scrapy_request, ScrapyRequest):
             raise TypeError('scrapy_request must be an instance of the ScrapyRequest class')
 
-        # Update the meta field with the meta field from the apify_request
-        meta = scrapy_request.meta or {}
-        meta.update({'apify_request_unique_key': apify_request.unique_key})
-        # scrapy_request.meta is a property, so we have to set it like this
-        scrapy_request._meta = meta  # noqa: SLF001
-
     # If the apify_request comes directly from the Scrapy, typically start URLs.
     else:
-        scrapy_request = ScrapyRequest(
-            url=apify_request.url,
-            method=apify_request.method,
-            meta={
-                'apify_request_unique_key': apify_request.unique_key,
-            },
-        )
+        scrapy_request = ScrapyRequest(url=apify_request.url, method=apify_request.method)
+
+    # Stamp the unique key with a fingerprint of the request it belongs to, so `to_apify_request` can tell
+    # this request apart from the ones Scrapy derives from it.
+    scrapy_request.meta['apify_request_unique_key'] = apify_request.unique_key
+    scrapy_request.meta['apify_request_fingerprint'] = _compute_fingerprint(scrapy_request)
 
     # Add optional 'headers' field
     if apify_request.headers: