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 it's 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 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,50 @@ def run_coro(
             future.cancel()
             raise
 
+    def submit_coro(self, coro: Coroutine) -> None:
+        """Schedule a coroutine on the event loop without waiting for its result.
+
+        Use this for work whose result nothing depends on, so the calling thread is not blocked by the round
+        trip. Failures are logged, as no caller is left 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.
+
+        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.')
+
+        # `wait_for_submitted` only runs once Scrapy goes idle, so without pruning here the list would hold
+        # every coroutine the whole crawl 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)
+
+    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 +160,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

@@ -76,8 +76,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']
+        # Reuse the queue's unique key only while this is still the request it was minted for. Redirects
+        # (`Request.replace()`) and spiders forwarding `meta` to another URL both inherit the stamp, and
+        # reusing it there deduplicates the derived request against its parent. A stamp without a URL beside
+        # it was set by hand, so it is taken at face value.
+        elif (unique_key := scrapy_request.meta.get('apify_request_unique_key')) and (
+            scrapy_request.meta.get('apify_request_url', scrapy_request.url) == scrapy_request.url
+        ):
+            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
@@ -187,21 +193,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 together with the URL 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_url'] = scrapy_request.url
 
     # Add optional 'headers' field
     if apify_request.headers:

src/apify/scrapy/scheduler.py

@@ -19,6 +19,8 @@
     from scrapy.http.request import Request
     from twisted.internet.defer import Deferred
 
+    from apify import Request as ApifyRequest
+
 logger = getLogger(__name__)
 
 
@@ -28,7 +30,11 @@ class ApifyScheduler(BaseScheduler):
     This scheduler requires the asyncio Twisted reactor to be installed.
     """
 
-    def __init__(self, async_thread_timeout: timedelta = timedelta(seconds=60)) -> None:
+    def __init__(
+        self,
+        async_thread_timeout: timedelta = timedelta(seconds=60),
+        crawler: Crawler | None = None,
+    ) -> None:
         if not is_asyncio_reactor_installed():
             raise ValueError(
                 f'{ApifyScheduler.__qualname__} requires the asyncio Twisted reactor. '
@@ -37,6 +43,10 @@ def __init__(self, async_thread_timeout: timedelta = timedelta(seconds=60)) -> N
             )
         self._rq: RequestQueue | None = None
         self.spider: Spider | None = None
+        self._crawler = crawler
+
+        self._requests_in_flight: list[tuple[ApifyRequest, Request]] = []
+        """Requests handed over to Scrapy and not resolved in the request queue yet."""
 
         # A thread with the asyncio event loop to run coroutines on.
         self._async_thread = AsyncThread(default_timeout=async_thread_timeout)
@@ -49,7 +59,7 @@ def from_crawler(cls, crawler: Crawler) -> ApifyScheduler:
         background event loop may take before timing out; it defaults to 60 seconds.
         """
         timeout_secs = crawler.settings.getint('APIFY_ASYNC_THREAD_TIMEOUT_SECS', 60)
-        return cls(async_thread_timeout=timedelta(seconds=timeout_secs))
+        return cls(async_thread_timeout=timedelta(seconds=timeout_secs), crawler=crawler)
 
     def open(self, spider: Spider) -> Deferred[None] | None:
         """Open the scheduler.
@@ -86,12 +96,34 @@ async def open_rq() -> RequestQueue:
     def close(self, reason: str) -> None:
         """Close the scheduler.
 
-        Shut down the event loop and its thread gracefully.
+        Resolve the requests Scrapy still holds, then shut down the event loop and its thread gracefully.
 
         Args:
             reason: The reason for closing the spider.
         """
         logger.debug(f'Closing {self.__class__.__name__} due to {reason}...')
+
+        rq = self._rq
+        if isinstance(rq, RequestQueue):
+            try:
+                self._resolve_finished_requests(wait=True)
+            except Exception:
+                logger.exception('Failed to resolve the requests still in flight in the request queue.')
+
+            # Whatever Scrapy did not finish goes back to the queue, so the next run gets it as pending.
+            # One failed reclaim must not strand the rest.
+            for apify_request, _ in self._requests_in_flight:
+                try:
+                    self._async_thread.run_coro(rq.reclaim_request(apify_request))
+                except Exception:
+                    logger.exception(f'Failed to reclaim the request {apify_request} in the request queue.')
+
+            self._requests_in_flight.clear()
+
+        # Closing the event loop cancels the updates fired off on the hot path silently, leaving those
+        # requests unhandled.
+        self._async_thread.wait_for_submitted()
+
         try:
             self._async_thread.close()
 
@@ -107,12 +139,21 @@ def close(self, reason: str) -> None:
     def has_pending_requests(self) -> bool:
         """Check if the scheduler has any pending requests.
 
+        Resolves the requests Scrapy has finished with first, as their outcome is what decides the answer.
+
         Returns:
             True if the scheduler has any pending requests, False otherwise.
         """
         if not isinstance(self._rq, RequestQueue):
             raise TypeError('self._rq must be an instance of the RequestQueue class')
 
+        # Scrapy asks this only once both its downloader and its scraper are idle, so everything still tracked
+        # as in flight is provably finished.
+        self._resolve_finished_requests(wait=True)
+
+        # The queue answers from its own bookkeeping, which a pending update has not reached yet.
+        self._async_thread.wait_for_submitted()
+
         # Log here before re-raising: this coroutine ran on a separate event-loop thread, and the failure is
         # otherwise easy to lose as it crosses that thread boundary back into Scrapy's synchronous machinery.
         try:
@@ -164,6 +205,10 @@ def next_request(self) -> Request | None:
         if not isinstance(self._rq, RequestQueue):
             raise TypeError('self._rq must be an instance of the RequestQueue class')
 
+        # The engine polls this method throughout the crawl, so resolving here keeps the queue current
+        # without blocking on the round trips.
+        self._resolve_finished_requests(wait=False)
+
         # Log here before re-raising: this coroutine ran on a separate event-loop thread, and the failure is
         # otherwise easy to lose as it crosses that thread boundary back into Scrapy's synchronous machinery.
         try:
@@ -178,26 +223,68 @@ def next_request(self) -> Request | None:
         if not isinstance(self.spider, Spider):
             raise TypeError('self.spider must be an instance of the Spider class')
 
-        # Reconstruct the Scrapy request before consuming the queue entry. A malformed entry must not crash
-        # the whole run, so on failure it is logged and skipped (None) rather than propagating.
+        # A corrupt or legacy payload must not crash the run, and is marked as handled right away, otherwise
+        # the queue would keep handing it back forever.
         try:
             scrapy_request = to_scrapy_request(apify_request, spider=self.spider)
         except Exception as exc:
             logger.warning(f'Failed to convert Apify request {apify_request} to a Scrapy request; skipping it: {exc}')
-            scrapy_request = None
-
-        # Mark the request as handled. This runs even when reconstruction failed above: an unrecoverable entry
-        # (a corrupt or legacy payload) must still be consumed, otherwise the queue would keep handing it back
-        # forever. Retrying genuine failures is the RetryMiddleware's job.
-        # Log here before re-raising: this coroutine ran on a separate event-loop thread, and the failure is
-        # otherwise easy to lose as it crosses that thread boundary back into Scrapy's synchronous machinery.
-        try:
-            self._async_thread.run_coro(self._rq.mark_request_as_handled(apify_request))
-        except Exception:
-            logger.exception('Failed to mark the request as handled in the request queue.')
-            raise
-
-        if scrapy_request is None:
+            try:
+                self._async_thread.run_coro(self._rq.mark_request_as_handled(apify_request))
+            except Exception:
+                logger.exception('Failed to mark the request as handled in the request queue.')
+                raise
             return None
 
+        # The entry stays unresolved until Scrapy is done with the request, so a run interrupted mid-flight
+        # leaves it pending instead of silently handled.
+        self._requests_in_flight.append((apify_request, scrapy_request))
+
         return scrapy_request
+
+    def _requests_busy_in_scrapy(self) -> set[Request]:
+        """Return the requests Scrapy is still working on.
+
+        A request joins the downloader's active set before the middleware chain runs and leaves the scraper's
+        only once the callback and the item pipeline are done, so absence from both means Scrapy has finished
+        with it - downloaded, dropped by a middleware or errored out alike.
+        """
+        engine = self._crawler.engine if self._crawler is not None else None
+        if engine is None:
+            return set()
+
+        scraper_slot = engine.scraper.slot
+        return engine.downloader.active | (scraper_slot.active if scraper_slot is not None else set())
+
+    def _resolve_finished_requests(self, *, wait: bool) -> None:
+        """Mark every request Scrapy has finished processing as handled in the request queue.
+
+        A request whose update cannot be dispatched stays tracked for the next call to retry, without holding
+        up the rest of the list.
+
+        Args:
+            wait: Whether to block until the queue has been updated. Pass False on the crawl's hot path, where
+                nothing depends on the result and blocking would stall the Twisted reactor.
+        """
+        rq = self._rq
+        if not self._requests_in_flight or not isinstance(rq, RequestQueue):
+            return
+
+        busy = self._requests_busy_in_scrapy()
+        unresolved: list[tuple[ApifyRequest, Request]] = []
+
+        for apify_request, scrapy_request in self._requests_in_flight:
+            if scrapy_request in busy:
+                unresolved.append((apify_request, scrapy_request))
+                continue
+
+            try:
+                if wait:
+                    self._async_thread.run_coro(rq.mark_request_as_handled(apify_request))
+                else:
+                    self._async_thread.submit_coro(rq.mark_request_as_handled(apify_request))
+            except Exception:
+                logger.exception(f'Failed to mark the request {apify_request} as handled in the request queue.')
+                unresolved.append((apify_request, scrapy_request))
+
+        self._requests_in_flight = unresolved

tests/unit/scrapy/requests/test_to_apify_request.py

@@ -10,6 +10,7 @@
 
 from crawlee._types import HttpHeaders
 
+from apify import Request as ApifyRequest
 from apify.scrapy.requests import to_apify_request, to_scrapy_request
 
 
@@ -187,3 +188,36 @@ def test_apify_request_id_in_meta_is_ignored(spider: Spider) -> None:
 
     assert apify_request is not None
     assert apify_request.unique_key == 'https://example.com'
+
+
+def test_unchanged_request_keeps_the_unique_key_it_was_stamped_with(spider: Spider) -> None:
+    """A request handed to Scrapy and enqueued again unchanged reuses the unique key it was minted for."""
+    scrapy_request = to_scrapy_request(ApifyRequest.from_url('https://example.com'), spider)
+
+    apify_request = to_apify_request(scrapy_request, spider)
+
+    assert apify_request is not None
+    assert apify_request.unique_key == scrapy_request.meta['apify_request_unique_key']
+
+
+def test_redirected_request_does_not_inherit_the_parents_unique_key(spider: Spider) -> None:
+    """A redirect derived from a fetched request gets its own unique key instead of the parent's stamp."""
+    parent = to_scrapy_request(ApifyRequest.from_url('https://example.com/redirect'), spider)
+    redirected = parent.replace(url='https://example.com/target')
+
+    apify_request = to_apify_request(redirected, spider)
+
+    assert apify_request is not None
+    assert apify_request.url == 'https://example.com/target'
+    assert apify_request.unique_key != parent.meta['apify_request_unique_key']
+
+
+def test_follow_up_request_with_propagated_meta_gets_its_own_unique_key(spider: Spider) -> None:
+    """A spider callback forwarding `meta` verbatim to another URL must not reuse the parent's unique key."""
+    parent = to_scrapy_request(ApifyRequest.from_url('https://example.com/listing'), spider)
+    follow_up = Request(url='https://example.com/detail', meta=parent.meta)
+
+    apify_request = to_apify_request(follow_up, spider)
+
+    assert apify_request is not None
+    assert apify_request.unique_key != parent.meta['apify_request_unique_key']