fix(scrapy): stop the crawl gracefully when the Actor run is migrated or aborted

docs/03_guides/06_scrapy.mdx

@@ -47,6 +47,7 @@ The Apify SDK provides several custom components to support integration with the
 - <ApiLink to="class/ActorDatasetPushPipeline">`apify.scrapy.pipelines.ActorDatasetPushPipeline`</ApiLink> - A Scrapy [item pipeline](https://docs.scrapy.org/en/latest/topics/item-pipeline.html) that pushes scraped items to Apify's [dataset](https://docs.apify.com/platform/storage/dataset). When enabled, every item produced by the spider is sent to the dataset.
 - <ApiLink to="class/ApifyHttpProxyMiddleware">`apify.scrapy.middlewares.ApifyHttpProxyMiddleware`</ApiLink> - A Scrapy [middleware](https://docs.scrapy.org/en/latest/topics/downloader-middleware.html) that manages proxy configurations. This middleware replaces Scrapy's default `HttpProxyMiddleware` to facilitate the use of Apify's proxy service.
 - <ApiLink to="class/ApifyCacheStorage">`apify.scrapy.extensions.ApifyCacheStorage`</ApiLink> - A storage backend for Scrapy's built-in [HTTP cache middleware](https://docs.scrapy.org/en/latest/topics/downloader-middleware.html#module-scrapy.downloadermiddlewares.httpcache). This backend uses Apify's [key-value store](https://docs.apify.com/platform/storage/key-value-store). To enable caching, set `HTTPCACHE_ENABLED` and `HTTPCACHE_EXPIRATION_SECS` in your settings. By default, when the spider closes, up to 100 expired and unreadable entries per run are cleaned up. To change this number, update `APIFY_HTTPCACHE_EXPIRATION_MAX_ITEMS`.
+- <ApiLink to="class/ApifyGracefulStopExtension">`apify.scrapy.extensions.ApifyGracefulStopExtension`</ApiLink> - A Scrapy [extension](https://docs.scrapy.org/en/latest/topics/extensions.html) that stops the crawl gracefully when the Actor run is aborted, so the requests in flight finish and get marked as handled. For details, see [Dealing with imminent migration to another host](#dealing-with-imminent-migration-to-another-host).
 
 Additional helper functions in the [`apify.scrapy`](https://github.com/apify/apify-sdk-python/tree/master/src/apify/scrapy) subpackage include:
 
@@ -104,9 +105,11 @@ 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 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.
+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. Before it does, it emits the `MIGRATING` [Actor event](../concepts/actor-events), and the integration reacts to it. The scheduler stops handing out requests to Scrapy, waits for the requests Scrapy is working on to finish, callbacks and item pipelines included, and marks them as handled in the request queue. The next run then continues with the pending requests instead of downloading the finished ones again and pushing their items a second time. Only the requests Scrapy hasn't finished when the platform kills the process are downloaded again. If you reboot the run with `Actor.reboot()`, the scheduler settles the requests in flight the same way before the reboot. The default `event_listeners_timeout` of 5 seconds may be too short for that, so pass a longer one.
 
-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.
+A graceful abort of the run works the same way. The `ApifyGracefulStopExtension` reacts to the `ABORTING` event by stopping the crawl: no new requests start, the requests in flight are marked as handled as they finish, and the spider closes once they all have. The requests still pending in the request queue stay there, so you can [resurrect](https://docs.apify.com/platform/actors/running/runs-and-builds#resurrection-of-finished-run) the run later.
+
+Note that the default `Spider.start()` yields the start URLs with `dont_filter=True`, which the integration maps to `always_enqueue=True`, so a restarted run crawls the start URLs again. To have them deduplicated against the request queue like any other request, override `start()` and yield plain requests, as the spider in [Example Actor](#example-actor) does. With `HTTPCACHE_ENABLED` and `HTTPCACHE_EXPIRATION_SECS` set, the requests a restarted run does download again hit the cache instead of the website.
 
 ## Conclusion
 

docs/03_guides/code/scrapy_project/src/spiders/title.py

@@ -8,7 +8,7 @@
 from ..items import TitleItem
 
 if TYPE_CHECKING:
-    from collections.abc import Generator
+    from collections.abc import AsyncIterator, Generator
 
     from scrapy.http.response import Response
 
@@ -33,6 +33,11 @@ def __init__(
         self.start_urls = start_urls
         self.allowed_domains = allowed_domains
 
+    async def start(self) -> AsyncIterator[Request]:
+        """Yield plain requests, so a restarted run doesn't crawl the start URLs again."""
+        for url in self.start_urls:
+            yield Request(url, callback=self.parse)
+
     def parse(self, response: Response) -> Generator[TitleItem | Request, None, None]:
         """Yield a `TitleItem` and a `Request` for each link on the page."""
         self.logger.info('TitleSpider is parsing %s...', response)

src/apify/scrapy/_warnings.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+import logging
+
+from crawlee._utils.log import LoggerOnce
+
+logger_once = LoggerOnce(logging.getLogger(__name__))
+"""Process-wide deduplication: the scheduler and the graceful-stop extension share the one message below."""
+
+
+def warn_about_uninitialized_actor() -> None:
+    """Warn, once per process, that the integration cannot react to the Actor run being migrated or aborted.
+
+    The scheduler and the graceful-stop extension both need an initialized Actor to listen for its events, and both
+    run without one in a plain `scrapy crawl`. One message covers them both.
+    """
+    logger_once.log(
+        'The Actor is not initialized, so the Scrapy integration cannot react to a migration or an abort of the '
+        'Actor run; the requests Scrapy is working on when the run is interrupted stay pending in the request queue.',
+        key='uninitialized-actor',
+        level=logging.WARNING,
+    )

src/apify/scrapy/extensions/__init__.py

@@ -1,3 +1,4 @@
+from apify.scrapy.extensions._graceful_stop import ApifyGracefulStopExtension
 from apify.scrapy.extensions._httpcache import ApifyCacheStorage
 
-__all__ = ['ApifyCacheStorage']
+__all__ = ['ApifyCacheStorage', 'ApifyGracefulStopExtension']

src/apify/scrapy/extensions/_graceful_stop.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from contextlib import suppress
+from logging import getLogger
+from typing import TYPE_CHECKING
+
+from scrapy import signals
+
+from apify import Actor, Event
+from apify.scrapy._warnings import warn_about_uninitialized_actor
+
+if TYPE_CHECKING:
+    from scrapy.crawler import Crawler
+
+logger = getLogger(__name__)
+
+
+class ApifyGracefulStopExtension:
+    """A Scrapy extension that stops the crawl gracefully when the Actor run is being aborted.
+
+    A graceful abort gives the run a moment before it is killed. The extension uses it to stop the engine: no new
+    requests are started, the ones in flight finish along with their callbacks and item pipelines, and the
+    scheduler marks them as handled in the request queue, so resurrecting the run does not download them again.
+    A migration is handled by `ApifyScheduler` instead, as the crawl must not finish on its own then.
+    """
+
+    def __init__(self, crawler: Crawler) -> None:
+        self._crawler = crawler
+
+    @classmethod
+    def from_crawler(cls, crawler: Crawler) -> ApifyGracefulStopExtension:
+        """Create the extension and hook it up to the spider's lifecycle."""
+        extension = cls(crawler)
+        crawler.signals.connect(extension.spider_opened, signal=signals.spider_opened)
+        crawler.signals.connect(extension.spider_closed, signal=signals.spider_closed)
+        return extension
+
+    def spider_opened(self) -> None:
+        """Start listening for the abort of the Actor run."""
+        try:
+            Actor.on(Event.ABORTING, self._on_aborting)
+        except RuntimeError:
+            warn_about_uninitialized_actor()
+
+    def spider_closed(self) -> None:
+        """Stop listening for the abort of the Actor run."""
+        # Without an initialized Actor (never initialized, or exited already) there is nothing to unregister from.
+        with suppress(RuntimeError):
+            Actor.off(Event.ABORTING, self._on_aborting)
+
+    async def _on_aborting(self) -> None:
+        """Stop the crawler; it waits for the requests in flight and closes the scheduler, which marks them."""
+        logger.info('The Actor run is being aborted: stopping the crawl gracefully.')
+        await self._crawler.stop_async()

src/apify/scrapy/scheduler.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+from contextlib import suppress
 from datetime import timedelta
 from logging import getLogger
 from typing import TYPE_CHECKING, Any
@@ -11,8 +12,9 @@
 from scrapy.utils.reactor import is_asyncio_reactor_installed
 
 from ._async_thread import AsyncThread
+from ._warnings import warn_about_uninitialized_actor
 from .requests import to_apify_request, to_scrapy_request
-from apify import Configuration
+from apify import Actor, Configuration, Event
 from apify.storage_clients import ApifyStorageClient
 from apify.storages import RequestQueue
 
@@ -24,13 +26,17 @@
     from scrapy.http.request import Request
     from twisted.internet.defer import Deferred
 
+    from apify import EventMigratingData
     from apify import Request as ApifyRequest
 
     InFlightRequest = tuple[ApifyRequest, Request]
     """A request handed over to Scrapy, paired with the RQ request it came from."""
 
 logger = getLogger(__name__)
 
+_SETTLE_POLL_INTERVAL = timedelta(seconds=1)
+"""How often the settling of a migration or an abort checks whether Scrapy has finished more of its requests."""
+
 
 async def _gather_failures(operations: Iterable[Coroutine[Any, Any, Any]]) -> list[BaseException | None]:
     """Run RQ updates concurrently, reporting each one's failure, or `None`, in the order given.
@@ -44,6 +50,11 @@ async def _gather_failures(operations: Iterable[Coroutine[Any, Any, Any]]) -> li
 class ApifyScheduler(BaseScheduler):
     """A Scrapy scheduler that uses the Apify `RequestQueue` to manage requests.
 
+    A request stays unresolved in the RQ until Scrapy is done with it, so an interrupted run leaves it pending for
+    the next one. When the platform is about to migrate the Actor run, the scheduler stops handing out requests;
+    then, as when the run is being aborted, it marks the ones Scrapy finishes as handled, so the next run does not
+    repeat them.
+
     This scheduler requires the asyncio Twisted reactor to be installed.
     """
 
@@ -68,6 +79,12 @@ def __init__(
         self._pending_marks: list[tuple[list[InFlightRequest], Future]] = []
         """Batches of mark-as-handled updates dispatched off the hot path, whose outcome is not known yet."""
 
+        self._migrating = False
+        """Whether the platform announced a migration of the Actor run; nothing is handed out to Scrapy then."""
+
+        self._closed = False
+        """Whether `close` has run; `_settle_requests_in_flight` stops then, as `close` resolves the rest itself."""
+
         # A thread with the asyncio event loop to run coroutines on.
         self._async_thread = AsyncThread(default_timeout=async_thread_timeout)
 
@@ -120,6 +137,12 @@ async def open_rq() -> RequestQueue:
                 logger.exception('Failed to close the async thread after a failed scheduler open.')
             raise
 
+        try:
+            Actor.on(Event.MIGRATING, self._on_migrating)
+            Actor.on(Event.ABORTING, self._on_aborting)
+        except RuntimeError:
+            warn_about_uninitialized_actor()
+
         return None
 
     def close(self, reason: str) -> None:
@@ -131,6 +154,12 @@ def close(self, reason: str) -> None:
             reason: The reason for closing the spider.
         """
         logger.debug(f'Closing {self.__class__.__name__} due to {reason}...')
+        self._closed = True
+
+        # Without an initialized Actor (never initialized, or exited already) there is nothing to unregister from.
+        with suppress(RuntimeError):
+            Actor.off(Event.MIGRATING, self._on_migrating)
+            Actor.off(Event.ABORTING, self._on_aborting)
 
         rq = self._rq
         if isinstance(rq, RequestQueue):
@@ -244,6 +273,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')
 
+        # Nothing goes out once a migration is announced; `_settle_requests_in_flight` handles what Scrapy holds.
+        if self._migrating:
+            return None
+
         # The engine polls this method throughout the crawl, so resolving here keeps the RQ current without
         # blocking on the round trips.
         self._resolve_finished_requests(wait=False)
@@ -281,6 +314,55 @@ def next_request(self) -> Request | None:
 
         return scrapy_request
 
+    async def _on_migrating(self, event_data: EventMigratingData) -> None:
+        """Stop handing out requests and settle the ones Scrapy holds, so the next run does not repeat them.
+
+        The platform restarts a migrating run on another host only if its process does not exit on its own: a
+        crawl that finished early would end the run as succeeded with work still pending. So the crawl is kept
+        running with nothing to do instead, and the requests Scrapy is still working on are marked as handled as
+        they finish, so the next run does not download them again and push their items a second time. Requests
+        still running when the process is killed stay pending.
+
+        Args:
+            event_data: The migration data; `time_remaining` tells how long until the process is killed.
+        """
+        if self._migrating:
+            return
+
+        self._migrating = True
+
+        remaining = event_data.time_remaining
+        deadline = '' if remaining is None else f' in {remaining.total_seconds():.0f} seconds'
+        logger.info(
+            f'The Actor run is migrating{deadline}: no more requests are handed out to Scrapy, and the '
+            f'{len(self._requests_in_flight)} request(s) it holds are marked as handled as they finish.'
+        )
+
+        await self._settle_requests_in_flight()
+
+    async def _on_aborting(self) -> None:
+        """Settle the requests Scrapy holds while the engine drains them, so a resurrected run does not repeat them.
+
+        `ApifyGracefulStopExtension` stops the engine, which closes the scheduler only once every request in flight
+        has finished. Until then nothing else marks the finished ones, so a request outliving the grace period of
+        the abort would leave all of them pending, their items already pushed.
+        """
+        logger.info(
+            f'The Actor run is being aborted: the {len(self._requests_in_flight)} request(s) Scrapy holds are marked '
+            'as handled as they finish.'
+        )
+
+        await self._settle_requests_in_flight()
+
+    async def _settle_requests_in_flight(self) -> None:
+        """Mark the requests Scrapy holds as handled as it finishes them, until none is left or `close` takes over."""
+        while not self._closed:
+            self._resolve_finished_requests(wait=True)
+            if not self._requests_in_flight:
+                logger.info('Scrapy has finished the requests it was working on.')
+                break
+            await asyncio.sleep(_SETTLE_POLL_INTERVAL.total_seconds())
+
     def _verify_engine_internals(self) -> None:
         """Fail at open time if Scrapy's engine no longer exposes what the in-flight tracking reads.
 

src/apify/scrapy/utils.py

@@ -47,6 +47,9 @@ def apply_apify_settings(*, settings: Settings | None = None, proxy_config: dict
     # Set the default HTTPCache middleware storage backend to ApifyCacheStorage
     settings['HTTPCACHE_STORAGE'] = 'apify.scrapy.extensions.ApifyCacheStorage'
 
+    # Stop the crawl gracefully when the Actor run is aborted
+    settings['EXTENSIONS']['apify.scrapy.extensions.ApifyGracefulStopExtension'] = 0
+
     # Store the proxy configuration
     settings['APIFY_PROXY_SETTINGS'] = proxy_config
 

tests/e2e/actor_source_base/server.py

@@ -16,6 +16,7 @@
     /redirect - Redirects (302) to /redirect-target
     /redirect-target - Page the redirect lands on
     /slow - Answers only after 10 minutes, to keep a request in flight while a run is interrupted
+    /delayed/<n> - Answers after 5 seconds with a link to /delayed/<n+1>, so a request is always in flight
 
 The homepage includes both direct product links (for Scrapy spiders that look for /products/ links
 on the start page) and category links (for testing crawl depth with Crawlee crawlers).
@@ -134,6 +135,20 @@ async def app(scope: dict[str, Any], _receive: Receive, send: Send) -> None:
     elif path == '/slow':
         await asyncio.sleep(600)
         await _send_html(send, '<html><head><title>Slow Page</title></head><body><h1>Slow Page</h1></body></html>')
+    elif path.startswith('/delayed/'):
+        try:
+            n = int(path.split('/')[-1])
+        except ValueError:
+            await _send_html(send, '<html><body>Not Found</body></html>', 404)
+            return
+        await asyncio.sleep(5)
+        await _send_html(
+            send,
+            f'<html><head><title>Delayed Page {n}</title></head><body>'
+            f'<h1>Delayed Page {n}</h1>'
+            f'<a href="/delayed/{n + 1}">Next</a>'
+            f'</body></html>',
+        )
     elif path.startswith('/deep/'):
         try:
             n = int(path.split('/')[-1])

tests/e2e/test_actor_scrapy.py

@@ -42,8 +42,11 @@ async def test_actor_scrapy_title_spider(
 
     items = await actor.last_run().dataset().list_items()
 
-    # CLOSESPIDER_PAGECOUNT is set to 10 in the spider settings.
-    assert items.count >= 9
+    # The start page and the pages it links to (`DEPTH_LIMIT` is 1, `CLOSESPIDER_PAGECOUNT` is 10), each scraped once.
+    urls = [item['url'] for item in items.items]
+    assert 'https://crawlee.dev' in urls
+    assert len(urls) > 1
+    assert len(urls) == len(set(urls)), urls
 
     for item in items.items:
         assert 'url' in item