From a83fe149db2a4f591c2f202eec1988e442531c5c Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 25 Aug 2026 15:00:45 +0200 Subject: [PATCH 1/9] fix(scrapy): stop the crawl gracefully when the Actor run is migrated or aborted --- docs/03_guides/06_scrapy.mdx | 7 +- .../code/scrapy_project/src/spiders/title.py | 7 +- src/apify/scrapy/extensions/__init__.py | 3 +- src/apify/scrapy/extensions/_graceful_stop.py | 55 ++++++++ src/apify/scrapy/scheduler.py | 71 +++++++++- src/apify/scrapy/utils.py | 3 + tests/e2e/actor_source_base/server.py | 15 +++ .../actor_source/spider_delayed_chain.py | 49 +++++++ .../test_scrapy/actor_source/spider_reboot.py | 74 ++++++++++ .../test_scrapy/test_graceful_abort_spider.py | 59 ++++++++ tests/e2e/test_scrapy/test_reboot_spider.py | 44 ++++++ .../scrapy/extensions/test_graceful_stop.py | 73 ++++++++++ tests/unit/scrapy/test_scheduler.py | 127 +++++++++++++++++- .../scrapy/utils/test_apply_apify_settings.py | 10 ++ 14 files changed, 585 insertions(+), 12 deletions(-) create mode 100644 src/apify/scrapy/extensions/_graceful_stop.py create mode 100644 tests/e2e/test_scrapy/actor_source/spider_delayed_chain.py create mode 100644 tests/e2e/test_scrapy/actor_source/spider_reboot.py create mode 100644 tests/e2e/test_scrapy/test_graceful_abort_spider.py create mode 100644 tests/e2e/test_scrapy/test_reboot_spider.py create mode 100644 tests/unit/scrapy/extensions/test_graceful_stop.py diff --git a/docs/03_guides/06_scrapy.mdx b/docs/03_guides/06_scrapy.mdx index 7e40633ad..91a770237 100644 --- a/docs/03_guides/06_scrapy.mdx +++ b/docs/03_guides/06_scrapy.mdx @@ -47,6 +47,7 @@ The Apify SDK provides several custom components to support integration with the - `apify.scrapy.pipelines.ActorDatasetPushPipeline` - 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. - `apify.scrapy.middlewares.ApifyHttpProxyMiddleware` - 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. - `apify.scrapy.extensions.ApifyCacheStorage` - 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`. +- `apify.scrapy.extensions.ApifyGracefulStopExtension` - 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 whose callbacks are still running when the platform kills the process are downloaded again. -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: the requests in flight finish and get marked as handled, and the spider closes. 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 diff --git a/docs/03_guides/code/scrapy_project/src/spiders/title.py b/docs/03_guides/code/scrapy_project/src/spiders/title.py index 8111ee31c..737db911c 100644 --- a/docs/03_guides/code/scrapy_project/src/spiders/title.py +++ b/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) diff --git a/src/apify/scrapy/extensions/__init__.py b/src/apify/scrapy/extensions/__init__.py index e9bccd1fa..0197e65f5 100644 --- a/src/apify/scrapy/extensions/__init__.py +++ b/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'] diff --git a/src/apify/scrapy/extensions/_graceful_stop.py b/src/apify/scrapy/extensions/_graceful_stop.py new file mode 100644 index 000000000..9e4c82c45 --- /dev/null +++ b/src/apify/scrapy/extensions/_graceful_stop.py @@ -0,0 +1,55 @@ +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 + +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: + logger.warning( + 'The Actor is not initialized, so the crawl cannot be stopped gracefully when the run is aborted.' + ) + + def spider_closed(self) -> None: + """Stop listening for the abort of the Actor run.""" + # The Actor may have exited already, in which case there is nothing left 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() diff --git a/src/apify/scrapy/scheduler.py b/src/apify/scrapy/scheduler.py index 0c4fb3968..a8a0e5acc 100644 --- a/src/apify/scrapy/scheduler.py +++ b/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 @@ -12,7 +13,7 @@ from ._async_thread import AsyncThread 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,6 +25,7 @@ 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] @@ -31,6 +33,9 @@ logger = getLogger(__name__) +SETTLE_POLL_INTERVAL = timedelta(seconds=1) +"""How often a migrating scheduler checks whether Scrapy has finished more of the requests it holds.""" + 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 +49,10 @@ 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 + and 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 +77,15 @@ 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._listening = False + """Whether `_on_migrating` is registered with the Actor, so `close` knows to unregister it.""" + + self._closed = False + """Whether `close` has run; `_on_migrating` stops settling requests then.""" + # A thread with the asyncio event loop to run coroutines on. self._async_thread = AsyncThread(default_timeout=async_thread_timeout) @@ -120,6 +138,16 @@ 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) + except RuntimeError: + logger.warning( + 'The Actor is not initialized, so the scheduler cannot react to a migration of the Actor run; the ' + 'requests Scrapy is working on when the run is interrupted stay pending in the request queue.' + ) + else: + self._listening = True + return None def close(self, reason: str) -> None: @@ -131,6 +159,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 + + if self._listening: + # The Actor may have exited already, in which case there is nothing left to unregister from. + with suppress(RuntimeError): + Actor.off(Event.MIGRATING, self._on_migrating) rq = self._rq if isinstance(rq, RequestQueue): @@ -244,6 +278,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; `_on_migrating` settles what Scrapy already 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 +319,37 @@ 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 is still working on are marked as handled as they finish.' + ) + + 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; waiting for the migration.') + 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. diff --git a/src/apify/scrapy/utils.py b/src/apify/scrapy/utils.py index 98af8d875..67f66979f 100644 --- a/src/apify/scrapy/utils.py +++ b/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 diff --git a/tests/e2e/actor_source_base/server.py b/tests/e2e/actor_source_base/server.py index 5299afaa2..04d46b97f 100644 --- a/tests/e2e/actor_source_base/server.py +++ b/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/ - Answers after 5 seconds with a link to /delayed/, 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, 'Slow Page

Slow Page

') + elif path.startswith('/delayed/'): + try: + n = int(path.split('/')[-1]) + except ValueError: + await _send_html(send, 'Not Found', 404) + return + await asyncio.sleep(5) + await _send_html( + send, + f'Delayed Page {n}' + f'

Delayed Page {n}

' + f'Next' + f'', + ) elif path.startswith('/deep/'): try: n = int(path.split('/')[-1]) diff --git a/tests/e2e/test_scrapy/actor_source/spider_delayed_chain.py b/tests/e2e/test_scrapy/actor_source/spider_delayed_chain.py new file mode 100644 index 000000000..074b784ba --- /dev/null +++ b/tests/e2e/test_scrapy/actor_source/spider_delayed_chain.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Any +from urllib.parse import urljoin + +from scrapy import Request, Spider + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Generator + + from scrapy.http.response import Response + +logger = logging.getLogger(__name__) + +IN_FLIGHT_LOG_MARKER = 'Handing the request over to the downloader' +"""Logged once Scrapy starts downloading a request, so a test knows the request is in flight.""" + + +class MarkInFlightMiddleware: + """Downloader middleware that logs `IN_FLIGHT_LOG_MARKER` for every request it sees.""" + + def process_request(self, request: Request) -> None: + logger.info(f'{IN_FLIGHT_LOG_MARKER}: {request.url}') + + +class DelayedChainSpider(Spider): + """Follow a chain of pages that each answer after a few seconds, so a request is always in flight.""" + + name = 'delayed_chain_spider' + + custom_settings = { # noqa: RUF012 + 'DOWNLOADER_MIDDLEWARES': {'src.spiders.spider.MarkInFlightMiddleware': 543}, + # One request at a time, so the test knows what is in flight when it interrupts the run. + 'CONCURRENT_REQUESTS': 1, + } + + def __init__(self, start_urls: list[str], *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.start_urls = start_urls + + async def start(self) -> AsyncIterator[Request]: + for url in self.start_urls: + yield Request(urljoin(url, '/delayed/1'), callback=self.parse) + + def parse(self, response: Response) -> Generator[dict | Request, None, None]: + yield {'url': response.url} + for href in response.css('a::attr("href")').getall(): + yield Request(urljoin(response.url, href), callback=self.parse) diff --git a/tests/e2e/test_scrapy/actor_source/spider_reboot.py b/tests/e2e/test_scrapy/actor_source/spider_reboot.py new file mode 100644 index 000000000..f3ca35d0c --- /dev/null +++ b/tests/e2e/test_scrapy/actor_source/spider_reboot.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +import asyncio +import logging +from datetime import timedelta +from typing import TYPE_CHECKING, Any +from urllib.parse import urljoin + +from scrapy import Request, Spider + +from apify import Actor + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Generator + + from scrapy.http.response import Response + +logger = logging.getLogger(__name__) + +REBOOT_LOG_MARKER = 'Rebooting the Actor run while a request is in flight' +"""Logged right before the reboot, so a test can tell the reboot happened.""" + +REBOOTED_KEY = 'REBOOTED' +"""Key-value store key set before the reboot, so the rebooted run does not reboot again.""" + +CHAIN_LENGTH = 3 +"""How many pages of the chain the spider follows.""" + + +class RebootOnceMiddleware: + """Downloader middleware that reboots the Actor once, while the first request it sees is being downloaded.""" + + def __init__(self) -> None: + self._reboot: asyncio.Future | None = None + + def process_request(self, request: Request) -> None: + if self._reboot is None: + logger.info(f'Handing the request over to the downloader: {request.url}') + self._reboot = asyncio.ensure_future(self._reboot_once()) + + async def _reboot_once(self) -> None: + if await Actor.get_value(REBOOTED_KEY): + return + + await Actor.set_value(REBOOTED_KEY, value=True) + logger.info(REBOOT_LOG_MARKER) + # The scheduler's migration listener first waits for the request in flight to finish, which takes a while. + await Actor.reboot(event_listeners_timeout=timedelta(seconds=30)) + + +class RebootSpider(Spider): + """Follow a short chain of slow pages, rebooting the Actor while the first one is being downloaded.""" + + name = 'reboot_spider' + + custom_settings = { # noqa: RUF012 + 'DOWNLOADER_MIDDLEWARES': {'src.spiders.spider.RebootOnceMiddleware': 543}, + 'CONCURRENT_REQUESTS': 1, + } + + def __init__(self, start_urls: list[str], *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.start_urls = start_urls + + async def start(self) -> AsyncIterator[Request]: + # Plain requests without `dont_filter`, so the rebooted run does not crawl the first page again. + for url in self.start_urls: + yield Request(urljoin(url, '/delayed/1'), callback=self.parse) + + def parse(self, response: Response) -> Generator[dict | Request, None, None]: + yield {'url': response.url} + if int(response.url.rsplit('/', 1)[-1]) < CHAIN_LENGTH: + for href in response.css('a::attr("href")').getall(): + yield Request(urljoin(response.url, href), callback=self.parse) diff --git a/tests/e2e/test_scrapy/test_graceful_abort_spider.py b/tests/e2e/test_scrapy/test_graceful_abort_spider.py new file mode 100644 index 000000000..5f318f73f --- /dev/null +++ b/tests/e2e/test_scrapy/test_graceful_abort_spider.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +from collections import Counter +from datetime import timedelta +from typing import TYPE_CHECKING + +from ..._utils import poll_until_condition +from .actor_source.spider_delayed_chain import IN_FLIGHT_LOG_MARKER +from .conftest import get_scrapy_source_files + +if TYPE_CHECKING: + from apify_client import ApifyClientAsync + + from ..conftest import MakeActorFunction + + +async def test_graceful_abort_spider(make_actor: MakeActorFunction, apify_client_async: ApifyClientAsync) -> None: + """A graceful abort lets the request in flight finish and marks it as handled, then the crawl closes on its own.""" + actor = await make_actor( + label='scrapy-graceful-abort', + source_files=get_scrapy_source_files('spider_delayed_chain.py', 'DelayedChainSpider'), + additional_requirements=['scrapy>=2.14.0'], + ) + run = await actor.start() + run_client = apify_client_async.run(run.id) + + # Abort the run only once Scrapy holds a request, otherwise there would be nothing to finish. The container + # startup time is highly variable, so poll the log with a growing interval. + log_client = run_client.log() + log = await poll_until_condition( + log_client.get, + lambda log: bool(log and IN_FLIGHT_LOG_MARKER in log), + timeout=300, + poll_interval=2, + backoff_factor=1.2, + ) + assert log is not None + assert IN_FLIGHT_LOG_MARKER in log, f'The run did not log {IN_FLIGHT_LOG_MARKER!r} in time:\n{log}' + + await run_client.abort(gracefully=True) + run_result = await run_client.wait_for_finish(wait_duration=timedelta(seconds=600)) + assert run_result is not None + assert run_result.status == 'ABORTED' + + # The crawl stopped gracefully before the platform killed the process. + log = await log_client.get() + assert log is not None + assert 'Spider closed (shutdown)' in log, f'The crawl did not close gracefully:\n{log}' + + # Every request Scrapy finished pushed its item exactly once, and the rest of the chain stays pending. + assert run.default_request_queue_id is not None + requests = await apify_client_async.request_queue(run.default_request_queue_id).list_requests() + handled_urls = [request.url for request in requests.items if request.handled_at is not None] + pending_urls = [request.url for request in requests.items if request.handled_at is None] + items = await run_client.dataset().list_items() + + assert handled_urls, f'No request was marked as handled: {requests.items}' + assert Counter(item['url'] for item in items.items) == Counter(handled_urls) + assert pending_urls, f'No request stayed pending: {requests.items}' diff --git a/tests/e2e/test_scrapy/test_reboot_spider.py b/tests/e2e/test_scrapy/test_reboot_spider.py new file mode 100644 index 000000000..6b98b1a9e --- /dev/null +++ b/tests/e2e/test_scrapy/test_reboot_spider.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from collections import Counter +from typing import TYPE_CHECKING + +from .actor_source.spider_reboot import CHAIN_LENGTH, REBOOT_LOG_MARKER +from .conftest import get_scrapy_source_files + +if TYPE_CHECKING: + from apify_client import ApifyClientAsync + + from ..conftest import MakeActorFunction, RunActorFunction + + +async def test_reboot_spider( + make_actor: MakeActorFunction, + run_actor: RunActorFunction, + apify_client_async: ApifyClientAsync, +) -> None: + """A run rebooted mid-crawl settles the request in flight first and continues where it left off afterwards.""" + actor = await make_actor( + label='scrapy-reboot', + source_files=get_scrapy_source_files('spider_reboot.py', 'RebootSpider'), + additional_requirements=['scrapy>=2.14.0'], + ) + run_result = await run_actor(actor) + assert run_result.status == 'SUCCEEDED' + + # The reboot went through the scheduler's migration handling before the process was restarted. + log = await actor.last_run().log().get() + assert log is not None + assert REBOOT_LOG_MARKER in log, f'The run did not reboot:\n{log}' + assert 'The Actor run is migrating' in log, f'The scheduler did not react to the reboot:\n{log}' + assert 'Scrapy has finished the requests it was working on' in log, f'The request in flight was not settled:\n{log}' + + # Each page of the chain was crawled once, the one in flight during the reboot included. + expected_urls = [f'http://localhost:8080/delayed/{n}' for n in range(1, CHAIN_LENGTH + 1)] + items = await actor.last_run().dataset().list_items() + assert sorted(item['url'] for item in items.items) == expected_urls + + assert run_result.default_request_queue_id is not None + requests = await apify_client_async.request_queue(run_result.default_request_queue_id).list_requests() + assert Counter(request.url for request in requests.items) == Counter(expected_urls) + assert all(request.handled_at is not None for request in requests.items), requests.items diff --git a/tests/unit/scrapy/extensions/test_graceful_stop.py b/tests/unit/scrapy/extensions/test_graceful_stop.py new file mode 100644 index 000000000..ca7f26bcf --- /dev/null +++ b/tests/unit/scrapy/extensions/test_graceful_stop.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +import logging +from types import SimpleNamespace +from typing import TYPE_CHECKING, Any +from unittest import mock + +from scrapy import signals +from scrapy.signalmanager import SignalManager + +from apify import Actor, Event, EventAbortingData +from apify.scrapy.extensions import ApifyGracefulStopExtension + +if TYPE_CHECKING: + import pytest + + +def fake_crawler() -> Any: + """Build a crawler stub with a real signal manager and a `stop_async` that records being awaited.""" + return SimpleNamespace(signals=SignalManager(), stop_async=mock.AsyncMock()) + + +def test_the_abort_listener_lives_as_long_as_the_spider(monkeypatch: pytest.MonkeyPatch) -> None: + """The abort listener is registered with the Actor when the spider opens and unregistered when it closes.""" + actor = mock.Mock() + monkeypatch.setattr('apify.scrapy.extensions._graceful_stop.Actor', actor) + crawler = fake_crawler() + extension = ApifyGracefulStopExtension.from_crawler(crawler) + + crawler.signals.send_catch_log(signal=signals.spider_opened, spider=None) + actor.on.assert_called_once_with(Event.ABORTING, extension._on_aborting) + actor.off.assert_not_called() + + crawler.signals.send_catch_log(signal=signals.spider_closed, spider=None, reason='finished') + actor.off.assert_called_once_with(Event.ABORTING, extension._on_aborting) + + +async def test_aborting_stops_the_crawler_gracefully() -> None: + """When the run is aborted, the crawler is stopped gracefully, so the requests in flight can finish.""" + crawler = fake_crawler() + extension = ApifyGracefulStopExtension.from_crawler(crawler) + + await extension._on_aborting() + + crawler.stop_async.assert_awaited_once_with() + + +async def test_the_actor_abort_event_reaches_the_extension() -> None: + """The abort event emitted by an initialized Actor stops the crawler, and no longer does once the spider closed.""" + crawler = fake_crawler() + extension = ApifyGracefulStopExtension.from_crawler(crawler) + + async with Actor: + extension.spider_opened() + Actor.event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) + await Actor.event_manager.wait_for_all_listeners_to_complete() + crawler.stop_async.assert_awaited_once_with() + + extension.spider_closed() + Actor.event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) + await Actor.event_manager.wait_for_all_listeners_to_complete() + crawler.stop_async.assert_awaited_once_with() + + +def test_spider_opened_warns_when_the_actor_is_not_initialized(caplog: pytest.LogCaptureFixture) -> None: + """Without an initialized Actor there is nothing to register the abort listener with, and that is said.""" + extension = ApifyGracefulStopExtension.from_crawler(fake_crawler()) + + with caplog.at_level(logging.WARNING, logger='apify.scrapy.extensions._graceful_stop'): + extension.spider_opened() + extension.spider_closed() + + assert 'Actor is not initialized' in caplog.text diff --git a/tests/unit/scrapy/test_scheduler.py b/tests/unit/scrapy/test_scheduler.py index df7fac9b6..d8f6b2665 100644 --- a/tests/unit/scrapy/test_scheduler.py +++ b/tests/unit/scrapy/test_scheduler.py @@ -2,7 +2,7 @@ import asyncio import logging -from concurrent.futures import Future +from concurrent.futures import Future, ThreadPoolExecutor from datetime import timedelta from types import SimpleNamespace from typing import TYPE_CHECKING, Any, cast @@ -12,6 +12,7 @@ from scrapy import Request, Spider from scrapy.settings import Settings +from apify import Event, EventMigratingData from apify import Request as ApifyRequest from apify.scrapy._async_thread import AsyncThread from apify.scrapy.scheduler import ApifyScheduler @@ -46,22 +47,27 @@ def fake_crawler( def fake_async_thread(default_timeout: timedelta | None = None) -> mock.Mock: # noqa: ARG001 - """Build an `AsyncThread` double that runs the scheduler's coroutines on a real event loop. + """Build an `AsyncThread` double that runs the scheduler's coroutines to completion on a real event loop. The scheduler batches the updates of a whole resolution pass into a single coroutine, so a double that - never runs them would leave these tests asserting on the batching instead of on what reaches the RQ. + never runs them would leave these tests asserting on the batching instead of on what reaches the RQ. Like + the real thread, the coroutines run on a worker thread, so the double also works from within an async test. """ + executor = ThreadPoolExecutor(max_workers=1) + + def run_coro(coro: Coroutine, timeout: Any = 'default') -> Any: # noqa: ARG001 + return executor.submit(asyncio.run, coro).result() def submit_coro(coro: Coroutine) -> Future: future: Future = Future() try: - future.set_result(asyncio.run(coro)) + future.set_result(run_coro(coro)) except Exception as exc: future.set_exception(exc) return future async_thread = mock.Mock(spec=AsyncThread) - async_thread.run_coro.side_effect = asyncio.run + async_thread.run_coro.side_effect = run_coro async_thread.submit_coro.side_effect = submit_coro return async_thread @@ -71,10 +77,15 @@ def called_methods(async_thread: mock.Mock) -> list[str]: return [name for name, _args, _kwargs in async_thread.method_calls] -def stub_scheduler_dependencies(monkeypatch: pytest.MonkeyPatch) -> None: - """Stub out the reactor check, the event loop thread and the RQ that `open` reaches for.""" +def stub_scheduler_dependencies(monkeypatch: pytest.MonkeyPatch) -> mock.Mock: + """Stub out the reactor check, the event loop thread, the Actor and the RQ that `open` reaches for. + + Returns the Actor double, for the tests that care about what the scheduler registers with it. + """ + actor = mock.Mock() monkeypatch.setattr('apify.scrapy.scheduler.is_asyncio_reactor_installed', lambda: True) monkeypatch.setattr('apify.scrapy.scheduler.AsyncThread', mock.Mock(side_effect=fake_async_thread)) + monkeypatch.setattr('apify.scrapy.scheduler.Actor', actor) async def open_rq(*_args: Any, **_kwargs: Any) -> Any: rq = mock.AsyncMock() @@ -83,6 +94,8 @@ async def open_rq(*_args: Any, **_kwargs: Any) -> Any: monkeypatch.setattr(RequestQueue, 'open', open_rq) + return actor + @pytest.fixture def scheduler(monkeypatch: pytest.MonkeyPatch, spider: DummySpider) -> ApifyScheduler: @@ -499,3 +512,103 @@ def test_from_crawler_keeps_the_crawler(monkeypatch: pytest.MonkeyPatch) -> None scheduler = ApifyScheduler.from_crawler(cast('Any', crawler)) assert scheduler._crawler is crawler + + +def test_open_listens_for_the_migration(monkeypatch: pytest.MonkeyPatch, spider: DummySpider) -> None: + """`open` registers the migration listener with the Actor, so the scheduler learns when the run is moving.""" + actor = stub_scheduler_dependencies(monkeypatch) + scheduler = ApifyScheduler(crawler=fake_crawler(scraper_busy=set())) + + scheduler.open(spider) + + actor.on.assert_called_once_with(Event.MIGRATING, scheduler._on_migrating) + + +def test_close_stops_listening_for_the_migration(monkeypatch: pytest.MonkeyPatch, spider: DummySpider) -> None: + """`close` unregisters the migration listener, so a closed scheduler is not told about a migration.""" + actor = stub_scheduler_dependencies(monkeypatch) + scheduler = ApifyScheduler(crawler=fake_crawler(scraper_busy=set())) + scheduler.open(spider) + + scheduler.close('finished') + + actor.off.assert_called_once_with(Event.MIGRATING, scheduler._on_migrating) + + +def test_open_warns_when_the_actor_is_not_initialized( + monkeypatch: pytest.MonkeyPatch, + spider: DummySpider, + caplog: pytest.LogCaptureFixture, +) -> None: + """Without an initialized Actor there is nothing to register the migration listener with, and that is said.""" + actor = stub_scheduler_dependencies(monkeypatch) + actor.on.side_effect = RuntimeError('The _ActorType is not active.') + scheduler = ApifyScheduler(crawler=fake_crawler(scraper_busy=set())) + + with caplog.at_level(logging.WARNING, logger='apify.scrapy.scheduler'): + scheduler.open(spider) + scheduler.close('finished') + + assert 'Actor is not initialized' in caplog.text + actor.off.assert_not_called() + + +async def test_migration_settles_the_requests_as_scrapy_finishes_them( + scheduler: ApifyScheduler, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Once a migration is announced, nothing more goes out and the requests Scrapy holds are marked as they finish.""" + monkeypatch.setattr('apify.scrapy.scheduler.SETTLE_POLL_INTERVAL', timedelta(milliseconds=10)) + rq = cast('mock.AsyncMock', scheduler._rq) + busy = set(hand_out(scheduler, 2)) + scheduler._crawler = fake_crawler(downloader_busy=busy) + (first_apify, first), (second_apify, second) = scheduler._requests_in_flight + rq.fetch_next_request.reset_mock() + + settled = asyncio.create_task(scheduler._on_migrating(EventMigratingData(time_remaining=timedelta(seconds=27)))) + await asyncio.sleep(0.05) + + rq.mark_request_as_handled.assert_not_called() + assert not settled.done() + + # The RQ is not even asked for more work. + rq.fetch_next_request.return_value = APIFY_REQUESTS[2] + assert scheduler.next_request() is None + rq.fetch_next_request.assert_not_called() + + busy.discard(first) + await asyncio.sleep(0.05) + rq.mark_request_as_handled.assert_called_once_with(first_apify) + assert not settled.done() + + busy.discard(second) + await asyncio.wait_for(settled, timeout=1) + assert rq.mark_request_as_handled.call_args_list == [mock.call(first_apify), mock.call(second_apify)] + + +async def test_migration_with_nothing_in_flight_only_stops_handing_out_requests(scheduler: ApifyScheduler) -> None: + """With nothing in flight the migration listener returns at once; the RQ is still not asked for more work.""" + rq = cast('mock.AsyncMock', scheduler._rq) + + await asyncio.wait_for(scheduler._on_migrating(EventMigratingData()), timeout=1) + + rq.fetch_next_request.return_value = APIFY_REQUESTS[0] + assert scheduler.next_request() is None + rq.fetch_next_request.assert_not_called() + + +async def test_close_ends_the_migration_settling(scheduler: ApifyScheduler, monkeypatch: pytest.MonkeyPatch) -> None: + """Closing the scheduler while it settles a migration ends the settling; `close` resolves the rest itself.""" + monkeypatch.setattr('apify.scrapy.scheduler.SETTLE_POLL_INTERVAL', timedelta(milliseconds=10)) + rq = cast('mock.AsyncMock', scheduler._rq) + hand_out(scheduler, 1) + + settled = asyncio.create_task(scheduler._on_migrating(EventMigratingData())) + await asyncio.sleep(0.05) + assert not settled.done() + + scheduler.close('shutdown') + await asyncio.wait_for(settled, timeout=1) + + rq.mark_request_as_handled.assert_not_called() + rq.reclaim_request.assert_called_once_with(APIFY_REQUESTS[0]) diff --git a/tests/unit/scrapy/utils/test_apply_apify_settings.py b/tests/unit/scrapy/utils/test_apply_apify_settings.py index 6c5227c02..c98f9164b 100644 --- a/tests/unit/scrapy/utils/test_apply_apify_settings.py +++ b/tests/unit/scrapy/utils/test_apply_apify_settings.py @@ -48,6 +48,16 @@ def test_updates_downloader_middlewares() -> None: } +def test_registers_graceful_stop_extension() -> None: + settings = Settings({'EXTENSIONS': {'scrapy.extensions.corestats.CoreStats': 500}}) + new_settings = apply_apify_settings(settings=settings) + + assert new_settings.get('EXTENSIONS') == { + 'scrapy.extensions.corestats.CoreStats': 500, + 'apify.scrapy.extensions.ApifyGracefulStopExtension': 0, + } + + def test_adds_proxy_config() -> None: settings = Settings() new_settings = apply_apify_settings(settings=settings) From faac5dba37dbd61cc89fd89e020406a9f6432c6c Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 25 Aug 2026 15:11:04 +0200 Subject: [PATCH 2/9] test(scrapy): replace casts in the scheduler tests with typed fixtures --- tests/unit/scrapy/test_scheduler.py | 186 +++++++++++++++------------- 1 file changed, 102 insertions(+), 84 deletions(-) diff --git a/tests/unit/scrapy/test_scheduler.py b/tests/unit/scrapy/test_scheduler.py index d8f6b2665..7a3b4a032 100644 --- a/tests/unit/scrapy/test_scheduler.py +++ b/tests/unit/scrapy/test_scheduler.py @@ -5,11 +5,12 @@ from concurrent.futures import Future, ThreadPoolExecutor from datetime import timedelta from types import SimpleNamespace -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, Any from unittest import mock import pytest from scrapy import Request, Spider +from scrapy.crawler import Crawler from scrapy.settings import Settings from apify import Event, EventMigratingData @@ -36,14 +37,23 @@ def fake_crawler( *, downloader_busy: set[Request] | None = None, scraper_busy: set[Request] | None = None, -) -> Any: - """Build a crawler stub reporting the given requests as busy; without `scraper_busy` its scraper slot is None.""" +) -> mock.Mock: + """Build a crawler double reporting the given requests as busy; without `scraper_busy` its scraper slot is None.""" scraper_slot = SimpleNamespace(active=scraper_busy) if scraper_busy is not None else None - engine = SimpleNamespace( + crawler = mock.Mock(spec=Crawler) + crawler.settings = Settings() + crawler.engine = SimpleNamespace( downloader=SimpleNamespace(active=downloader_busy if downloader_busy is not None else set()), scraper=SimpleNamespace(slot=scraper_slot), ) - return SimpleNamespace(engine=engine) + return crawler + + +def fake_rq() -> mock.AsyncMock: + """Build an RQ double that passes the scheduler's `isinstance` check.""" + rq = mock.AsyncMock() + rq.__class__ = RequestQueue + return rq def fake_async_thread(default_timeout: timedelta | None = None) -> mock.Mock: # noqa: ARG001 @@ -87,10 +97,8 @@ def stub_scheduler_dependencies(monkeypatch: pytest.MonkeyPatch) -> mock.Mock: monkeypatch.setattr('apify.scrapy.scheduler.AsyncThread', mock.Mock(side_effect=fake_async_thread)) monkeypatch.setattr('apify.scrapy.scheduler.Actor', actor) - async def open_rq(*_args: Any, **_kwargs: Any) -> Any: - rq = mock.AsyncMock() - rq.__class__ = RequestQueue - return rq + async def open_rq(*_args: Any, **_kwargs: Any) -> RequestQueue: + return fake_rq() monkeypatch.setattr(RequestQueue, 'open', open_rq) @@ -98,24 +106,37 @@ async def open_rq(*_args: Any, **_kwargs: Any) -> Any: @pytest.fixture -def scheduler(monkeypatch: pytest.MonkeyPatch, spider: DummySpider) -> ApifyScheduler: +def rq() -> mock.AsyncMock: + """The RQ double the `scheduler` fixture talks to.""" + return fake_rq() + + +@pytest.fixture +def async_thread() -> mock.Mock: + """The `AsyncThread` double the `scheduler` fixture runs its coroutines on.""" + return fake_async_thread() + + +@pytest.fixture +def scheduler( + monkeypatch: pytest.MonkeyPatch, + spider: DummySpider, + rq: mock.AsyncMock, + async_thread: mock.Mock, +) -> ApifyScheduler: """Create a scheduler with its reactor check stubbed out, a fake event loop thread and a mocked RQ.""" stub_scheduler_dependencies(monkeypatch) + monkeypatch.setattr('apify.scrapy.scheduler.AsyncThread', mock.Mock(return_value=async_thread)) scheduler = ApifyScheduler() scheduler.spider = spider - - rq = mock.AsyncMock() - rq.__class__ = RequestQueue scheduler._rq = rq return scheduler -def test_has_pending_requests_reflects_queue_state(scheduler: ApifyScheduler) -> None: +def test_has_pending_requests_reflects_queue_state(scheduler: ApifyScheduler, rq: mock.AsyncMock) -> None: """`has_pending_requests` is True while the queue is not finished and False once it is.""" - rq = cast('mock.AsyncMock', scheduler._rq) - rq.is_finished.return_value = False # the RQ still has work assert scheduler.has_pending_requests() is True @@ -126,10 +147,9 @@ def test_has_pending_requests_reflects_queue_state(scheduler: ApifyScheduler) -> def test_enqueue_request_skips_non_serializable_request( scheduler: ApifyScheduler, caplog: pytest.LogCaptureFixture, + rq: mock.AsyncMock, ) -> None: """A request that cannot be converted (non-serializable meta) is not enqueued: returns False and logs a warning.""" - rq = cast('mock.MagicMock', scheduler._rq) - # A set in `meta` is not JSON-serializable, so `to_apify_request` returns None. scrapy_request = Request(url='https://example.com', meta={'tags': {'a', 'b'}}) @@ -141,9 +161,8 @@ def test_enqueue_request_skips_non_serializable_request( rq.add_request.assert_not_called() -def test_enqueue_request_enqueues_converted_request(scheduler: ApifyScheduler) -> None: +def test_enqueue_request_enqueues_converted_request(scheduler: ApifyScheduler, rq: mock.AsyncMock) -> None: """A convertible request is enqueued and reported as newly added when the queue had not seen it.""" - rq = cast('mock.AsyncMock', scheduler._rq) rq.add_request.return_value = SimpleNamespace(was_already_present=False) result = scheduler.enqueue_request(Request(url='https://example.com')) @@ -152,9 +171,8 @@ def test_enqueue_request_enqueues_converted_request(scheduler: ApifyScheduler) - rq.add_request.assert_called_once() -def test_enqueue_request_returns_false_for_duplicate(scheduler: ApifyScheduler) -> None: +def test_enqueue_request_returns_false_for_duplicate(scheduler: ApifyScheduler, rq: mock.AsyncMock) -> None: """A request already present in the queue is reported as not newly enqueued (returns False).""" - rq = cast('mock.AsyncMock', scheduler._rq) rq.add_request.return_value = SimpleNamespace(was_already_present=True) result = scheduler.enqueue_request(Request(url='https://example.com')) @@ -165,10 +183,9 @@ def test_enqueue_request_returns_false_for_duplicate(scheduler: ApifyScheduler) def test_next_request_skips_request_that_fails_to_convert( scheduler: ApifyScheduler, caplog: pytest.LogCaptureFixture, + rq: mock.AsyncMock, ) -> None: """A queue entry that fails to reconstruct is skipped and still marked handled, not retried forever.""" - rq = cast('mock.AsyncMock', scheduler._rq) - # A queue entry whose encoded Scrapy request is malformed; `to_scrapy_request` raises on it. malformed_request = ApifyRequest( url='https://example.com', @@ -191,10 +208,8 @@ def test_next_request_skips_request_that_fails_to_convert( rq.mark_request_as_handled.assert_called_once_with(malformed_request) -def test_next_request_returns_converted_request(scheduler: ApifyScheduler) -> None: +def test_next_request_returns_converted_request(scheduler: ApifyScheduler, rq: mock.AsyncMock) -> None: """A valid RQ entry is reconstructed into a Scrapy request and left unhandled until Scrapy is done.""" - rq = cast('mock.AsyncMock', scheduler._rq) - apify_request = ApifyRequest( url='https://example.com', method='GET', @@ -210,9 +225,8 @@ def test_next_request_returns_converted_request(scheduler: ApifyScheduler) -> No rq.mark_request_as_handled.assert_not_called() -def test_next_request_returns_none_when_queue_empty(scheduler: ApifyScheduler) -> None: +def test_next_request_returns_none_when_queue_empty(scheduler: ApifyScheduler, rq: mock.AsyncMock) -> None: """An empty queue makes `next_request` return None and skip marking anything as handled.""" - rq = cast('mock.AsyncMock', scheduler._rq) rq.fetch_next_request.return_value = None result = scheduler.next_request() @@ -224,9 +238,9 @@ def test_next_request_returns_none_when_queue_empty(scheduler: ApifyScheduler) - def test_next_request_logs_exception_before_propagating( scheduler: ApifyScheduler, caplog: pytest.LogCaptureFixture, + rq: mock.AsyncMock, ) -> None: """A failure in the coroutine run is logged with its traceback via `logger.exception` before propagating.""" - rq = cast('mock.AsyncMock', scheduler._rq) rq.fetch_next_request.side_effect = RuntimeError('boom') with caplog.at_level(logging.ERROR, logger='apify.scrapy.scheduler'), pytest.raises(RuntimeError, match='boom'): @@ -246,8 +260,9 @@ def test_from_crawler_reads_async_thread_timeout_setting(monkeypatch: pytest.Mon async_thread_cls = mock.Mock(side_effect=fake_async_thread) monkeypatch.setattr('apify.scrapy.scheduler.AsyncThread', async_thread_cls) - crawler = SimpleNamespace(settings=Settings({'APIFY_ASYNC_THREAD_TIMEOUT_SECS': 123})) - ApifyScheduler.from_crawler(cast('Any', crawler)) + crawler = fake_crawler() + crawler.settings = Settings({'APIFY_ASYNC_THREAD_TIMEOUT_SECS': 123}) + ApifyScheduler.from_crawler(crawler) async_thread_cls.assert_called_once_with(default_timeout=timedelta(seconds=123)) @@ -263,19 +278,20 @@ def test_from_crawler_reads_async_thread_timeout_setting(monkeypatch: pytest.Mon ] -def hand_out(scheduler: ApifyScheduler, count: int) -> list[Request]: +def hand_out(scheduler: ApifyScheduler, rq: mock.AsyncMock, count: int) -> list[Request]: """Fetch `count` requests, keeping each busy in Scrapy so it is not resolved as it is handed over. - Returns the Scrapy requests, and leaves the crawler stub reporting all of them as busy in the downloader. + Returns the Scrapy requests, and leaves the crawler double reporting all of them as busy in the downloader. """ - rq = cast('mock.AsyncMock', scheduler._rq) rq.fetch_next_request.side_effect = APIFY_REQUESTS[:count] busy: set[Request] = set() scheduler._crawler = fake_crawler(downloader_busy=busy) for _ in range(count): - busy.add(cast('Request', scheduler.next_request())) + scrapy_request = scheduler.next_request() + assert scrapy_request is not None + busy.add(scrapy_request) rq.fetch_next_request.side_effect = None rq.fetch_next_request.return_value = None @@ -283,10 +299,8 @@ def hand_out(scheduler: ApifyScheduler, count: int) -> list[Request]: return list(busy) -def test_has_pending_requests_marks_finished_requests_as_handled(scheduler: ApifyScheduler) -> None: +def test_has_pending_requests_marks_finished_requests_as_handled(scheduler: ApifyScheduler, rq: mock.AsyncMock) -> None: """Requests Scrapy has finished with are marked as handled once it goes idle and asks about pending work.""" - rq = cast('mock.AsyncMock', scheduler._rq) - rq.fetch_next_request.return_value = APIFY_REQUESTS[0] scheduler.next_request() rq.mark_request_as_handled.assert_not_called() @@ -299,12 +313,13 @@ def test_has_pending_requests_marks_finished_requests_as_handled(scheduler: Apif rq.mark_request_as_handled.assert_called_once_with(APIFY_REQUESTS[0]) -def test_next_request_marks_finished_requests_without_blocking(scheduler: ApifyScheduler) -> None: +def test_next_request_marks_finished_requests_without_blocking( + scheduler: ApifyScheduler, + rq: mock.AsyncMock, + async_thread: mock.Mock, +) -> None: """On the crawl's hot path a finished request is marked as handled without blocking the reactor on it.""" - rq = cast('mock.AsyncMock', scheduler._rq) - async_thread = cast('mock.Mock', scheduler._async_thread) - - (scrapy_request,) = hand_out(scheduler, 1) + (scrapy_request,) = hand_out(scheduler, rq, 1) # Scrapy is still downloading the request, so it stays unresolved. scheduler._crawler = fake_crawler(downloader_busy={scrapy_request}) @@ -320,10 +335,12 @@ def test_next_request_marks_finished_requests_without_blocking(scheduler: ApifyS assert 'submit_coro' in called_methods(async_thread) -def test_has_pending_requests_waits_for_the_non_blocking_updates(scheduler: ApifyScheduler) -> None: +def test_has_pending_requests_waits_for_the_non_blocking_updates( + scheduler: ApifyScheduler, + rq: mock.AsyncMock, + async_thread: mock.Mock, +) -> None: """The RQ is asked whether it is finished only after the updates fired off on the hot path have landed.""" - rq = cast('mock.AsyncMock', scheduler._rq) - async_thread = cast('mock.Mock', scheduler._async_thread) scheduler._crawler = fake_crawler() rq.is_finished.return_value = True @@ -340,11 +357,13 @@ def test_has_pending_requests_waits_for_the_non_blocking_updates(scheduler: Apif pytest.param('scraper_busy', id='busy in the scraper slot'), ], ) -def test_close_reclaims_requests_scrapy_never_finished(scheduler: ApifyScheduler, busy_kwarg: str) -> None: +def test_close_reclaims_requests_scrapy_never_finished( + scheduler: ApifyScheduler, + busy_kwarg: str, + rq: mock.AsyncMock, +) -> None: """Requests still being processed when the scheduler closes go back to the RQ instead of being lost.""" - rq = cast('mock.AsyncMock', scheduler._rq) - - (scrapy_request,) = hand_out(scheduler, 1) + (scrapy_request,) = hand_out(scheduler, rq, 1) # Scrapy is still working on the request when the run is interrupted. scheduler._crawler = fake_crawler(**{busy_kwarg: {scrapy_request}}) @@ -355,11 +374,9 @@ def test_close_reclaims_requests_scrapy_never_finished(scheduler: ApifyScheduler rq.mark_request_as_handled.assert_not_called() -def test_close_marks_the_requests_scrapy_finished_as_handled(scheduler: ApifyScheduler) -> None: +def test_close_marks_the_requests_scrapy_finished_as_handled(scheduler: ApifyScheduler, rq: mock.AsyncMock) -> None: """Requests Scrapy drained before the shutdown are marked as handled rather than reclaimed.""" - rq = cast('mock.AsyncMock', scheduler._rq) - - hand_out(scheduler, 1) + hand_out(scheduler, rq, 1) # Scrapy drains its downloader and its scraper before the scheduler is closed. scheduler._crawler = fake_crawler(scraper_busy=set()) @@ -373,11 +390,10 @@ def test_close_marks_the_requests_scrapy_finished_as_handled(scheduler: ApifySch def test_close_reclaims_the_other_requests_after_a_failed_reclaim( scheduler: ApifyScheduler, caplog: pytest.LogCaptureFixture, + rq: mock.AsyncMock, ) -> None: """One failing reclaim does not stop the other in-flight requests from going back to the RQ.""" - rq = cast('mock.AsyncMock', scheduler._rq) - - hand_out(scheduler, 2) + hand_out(scheduler, rq, 2) rq.reclaim_request.side_effect = [RuntimeError('boom'), None] with caplog.at_level(logging.ERROR, logger='apify.scrapy.scheduler'): @@ -388,12 +404,13 @@ def test_close_reclaims_the_other_requests_after_a_failed_reclaim( assert len(errors) == 1 -def test_close_reaches_the_rq_in_one_round_trip_per_operation(scheduler: ApifyScheduler) -> None: +def test_close_reaches_the_rq_in_one_round_trip_per_operation( + scheduler: ApifyScheduler, + rq: mock.AsyncMock, + async_thread: mock.Mock, +) -> None: """Marks and reclaims each travel together, as a migration may not leave room for one round trip each.""" - rq = cast('mock.AsyncMock', scheduler._rq) - async_thread = cast('mock.Mock', scheduler._async_thread) - - handed_out = hand_out(scheduler, 4) + handed_out = hand_out(scheduler, rq, 4) # Scrapy finished half of the requests and is still working on the rest when the run is interrupted. scheduler._crawler = fake_crawler(downloader_busy=set(handed_out[2:])) @@ -406,10 +423,8 @@ def test_close_reaches_the_rq_in_one_round_trip_per_operation(scheduler: ApifySc assert called_methods(async_thread).count('run_coro') == 2 -def test_close_waits_for_the_non_blocking_updates(scheduler: ApifyScheduler) -> None: +def test_close_waits_for_the_non_blocking_updates(scheduler: ApifyScheduler, async_thread: mock.Mock) -> None: """The event loop is not torn down before the updates fired off on the hot path have landed.""" - async_thread = cast('mock.Mock', scheduler._async_thread) - scheduler.close('finished') methods = called_methods(async_thread) @@ -419,11 +434,10 @@ def test_close_waits_for_the_non_blocking_updates(scheduler: ApifyScheduler) -> def test_a_failed_mark_keeps_the_request_tracked( scheduler: ApifyScheduler, caplog: pytest.LogCaptureFixture, + rq: mock.AsyncMock, ) -> None: """A request whose mark-as-handled fails stays tracked, so the next resolution retries it.""" - rq = cast('mock.AsyncMock', scheduler._rq) - - hand_out(scheduler, 1) + hand_out(scheduler, rq, 1) scheduler._crawler = fake_crawler() # The mark fails, then the RQ reports itself unfinished because the request is still in progress. @@ -441,11 +455,10 @@ def test_a_failed_mark_keeps_the_request_tracked( def test_a_mark_that_fails_after_being_fired_off_is_retried( scheduler: ApifyScheduler, caplog: pytest.LogCaptureFixture, + rq: mock.AsyncMock, ) -> None: """A hot-path mark that fails on its way to the RQ is retried, instead of leaving the request in progress.""" - rq = cast('mock.AsyncMock', scheduler._rq) - - hand_out(scheduler, 1) + hand_out(scheduler, rq, 1) # Scrapy is done with the request, so the hot path fires the mark off - and it fails out of sight. scheduler._crawler = fake_crawler() @@ -498,18 +511,19 @@ def test_open_fails_loudly_when_the_scrapy_engine_internals_move( ) -> None: """A Scrapy release that moves the engine internals has to break at open, not silently mid-crawl.""" stub_scheduler_dependencies(monkeypatch) - crawler = SimpleNamespace(engine=SimpleNamespace(downloader=SimpleNamespace(), scraper=SimpleNamespace())) + crawler = fake_crawler() + crawler.engine = SimpleNamespace(downloader=SimpleNamespace(), scraper=SimpleNamespace()) with pytest.raises(RuntimeError, match='engine internals'): - ApifyScheduler(crawler=cast('Any', crawler)).open(spider) + ApifyScheduler(crawler=crawler).open(spider) def test_from_crawler_keeps_the_crawler(monkeypatch: pytest.MonkeyPatch) -> None: """`from_crawler` keeps the crawler, which is how the scheduler learns what Scrapy is still working on.""" stub_scheduler_dependencies(monkeypatch) - crawler = SimpleNamespace(settings=Settings()) - scheduler = ApifyScheduler.from_crawler(cast('Any', crawler)) + crawler = fake_crawler() + scheduler = ApifyScheduler.from_crawler(crawler) assert scheduler._crawler is crawler @@ -556,11 +570,11 @@ def test_open_warns_when_the_actor_is_not_initialized( async def test_migration_settles_the_requests_as_scrapy_finishes_them( scheduler: ApifyScheduler, monkeypatch: pytest.MonkeyPatch, + rq: mock.AsyncMock, ) -> None: """Once a migration is announced, nothing more goes out and the requests Scrapy holds are marked as they finish.""" monkeypatch.setattr('apify.scrapy.scheduler.SETTLE_POLL_INTERVAL', timedelta(milliseconds=10)) - rq = cast('mock.AsyncMock', scheduler._rq) - busy = set(hand_out(scheduler, 2)) + busy = set(hand_out(scheduler, rq, 2)) scheduler._crawler = fake_crawler(downloader_busy=busy) (first_apify, first), (second_apify, second) = scheduler._requests_in_flight rq.fetch_next_request.reset_mock() @@ -586,10 +600,11 @@ async def test_migration_settles_the_requests_as_scrapy_finishes_them( assert rq.mark_request_as_handled.call_args_list == [mock.call(first_apify), mock.call(second_apify)] -async def test_migration_with_nothing_in_flight_only_stops_handing_out_requests(scheduler: ApifyScheduler) -> None: +async def test_migration_with_nothing_in_flight_only_stops_handing_out_requests( + scheduler: ApifyScheduler, + rq: mock.AsyncMock, +) -> None: """With nothing in flight the migration listener returns at once; the RQ is still not asked for more work.""" - rq = cast('mock.AsyncMock', scheduler._rq) - await asyncio.wait_for(scheduler._on_migrating(EventMigratingData()), timeout=1) rq.fetch_next_request.return_value = APIFY_REQUESTS[0] @@ -597,11 +612,14 @@ async def test_migration_with_nothing_in_flight_only_stops_handing_out_requests( rq.fetch_next_request.assert_not_called() -async def test_close_ends_the_migration_settling(scheduler: ApifyScheduler, monkeypatch: pytest.MonkeyPatch) -> None: +async def test_close_ends_the_migration_settling( + scheduler: ApifyScheduler, + monkeypatch: pytest.MonkeyPatch, + rq: mock.AsyncMock, +) -> None: """Closing the scheduler while it settles a migration ends the settling; `close` resolves the rest itself.""" monkeypatch.setattr('apify.scrapy.scheduler.SETTLE_POLL_INTERVAL', timedelta(milliseconds=10)) - rq = cast('mock.AsyncMock', scheduler._rq) - hand_out(scheduler, 1) + hand_out(scheduler, rq, 1) settled = asyncio.create_task(scheduler._on_migrating(EventMigratingData())) await asyncio.sleep(0.05) From 20e930f67ab2bf3e523f53f4ba9cc4870bd004a6 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 25 Aug 2026 15:18:00 +0200 Subject: [PATCH 3/9] test(scrapy): assert unique items instead of a count inflated by a duplicate start page --- tests/e2e/test_actor_scrapy.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/tests/e2e/test_actor_scrapy.py b/tests/e2e/test_actor_scrapy.py index 363b84e7f..5c04c2c89 100644 --- a/tests/e2e/test_actor_scrapy.py +++ b/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 in the project settings), 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 From 404ad8d36adfe30824fc0ad9d9e9956d5a3211ef Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 25 Aug 2026 19:32:23 +0200 Subject: [PATCH 4/9] fix: mark in-flight Scrapy requests as handled while a graceful abort drains them --- docs/03_guides/06_scrapy.mdx | 2 +- src/apify/scrapy/scheduler.py | 39 ++++++++++++---- tests/unit/scrapy/test_scheduler.py | 69 ++++++++++++++++++++++++++--- 3 files changed, 93 insertions(+), 17 deletions(-) diff --git a/docs/03_guides/06_scrapy.mdx b/docs/03_guides/06_scrapy.mdx index 91a770237..08051915a 100644 --- a/docs/03_guides/06_scrapy.mdx +++ b/docs/03_guides/06_scrapy.mdx @@ -107,7 +107,7 @@ The following example shows a Scrapy Actor that scrapes page titles and enqueues 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 whose callbacks are still running when the platform kills the process are downloaded again. -A graceful abort of the run works the same way. The `ApifyGracefulStopExtension` reacts to the `ABORTING` event by stopping the crawl: the requests in flight finish and get marked as handled, and the spider closes. 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. +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. diff --git a/src/apify/scrapy/scheduler.py b/src/apify/scrapy/scheduler.py index a8a0e5acc..9fb9b794d 100644 --- a/src/apify/scrapy/scheduler.py +++ b/src/apify/scrapy/scheduler.py @@ -34,7 +34,7 @@ logger = getLogger(__name__) SETTLE_POLL_INTERVAL = timedelta(seconds=1) -"""How often a migrating scheduler checks whether Scrapy has finished more of the requests it holds.""" +"""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]: @@ -50,8 +50,9 @@ 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 - and marks the ones Scrapy finishes as handled, so the next run does not repeat them. + 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. """ @@ -84,7 +85,7 @@ def __init__( """Whether `_on_migrating` is registered with the Actor, so `close` knows to unregister it.""" self._closed = False - """Whether `close` has run; `_on_migrating` stops settling requests then.""" + """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) @@ -140,10 +141,11 @@ async def open_rq() -> RequestQueue: try: Actor.on(Event.MIGRATING, self._on_migrating) + Actor.on(Event.ABORTING, self._on_aborting) except RuntimeError: logger.warning( - 'The Actor is not initialized, so the scheduler cannot react to a migration of the Actor run; the ' - 'requests Scrapy is working on when the run is interrupted stay pending in the request queue.' + 'The Actor is not initialized, so the scheduler 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.' ) else: self._listening = True @@ -165,6 +167,7 @@ def close(self, reason: str) -> None: # The Actor may have exited already, in which case there is nothing left 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): @@ -278,7 +281,7 @@ 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; `_on_migrating` settles what Scrapy already holds. + # Nothing goes out once a migration is announced; `_settle_requests_in_flight` handles what Scrapy holds. if self._migrating: return None @@ -340,13 +343,31 @@ async def _on_migrating(self, event_data: EventMigratingData) -> None: 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 is still working on are marked as handled as they finish.' + 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; waiting for the migration.') + logger.info('Scrapy has finished the requests it was working on.') break await asyncio.sleep(SETTLE_POLL_INTERVAL.total_seconds()) diff --git a/tests/unit/scrapy/test_scheduler.py b/tests/unit/scrapy/test_scheduler.py index 7a3b4a032..3f1920c55 100644 --- a/tests/unit/scrapy/test_scheduler.py +++ b/tests/unit/scrapy/test_scheduler.py @@ -528,25 +528,34 @@ def test_from_crawler_keeps_the_crawler(monkeypatch: pytest.MonkeyPatch) -> None assert scheduler._crawler is crawler -def test_open_listens_for_the_migration(monkeypatch: pytest.MonkeyPatch, spider: DummySpider) -> None: - """`open` registers the migration listener with the Actor, so the scheduler learns when the run is moving.""" +def test_open_listens_for_the_migration_and_the_abort(monkeypatch: pytest.MonkeyPatch, spider: DummySpider) -> None: + """`open` registers the listeners with the Actor, so the scheduler learns when the run is moving or aborting.""" actor = stub_scheduler_dependencies(monkeypatch) scheduler = ApifyScheduler(crawler=fake_crawler(scraper_busy=set())) scheduler.open(spider) - actor.on.assert_called_once_with(Event.MIGRATING, scheduler._on_migrating) + assert actor.on.call_args_list == [ + mock.call(Event.MIGRATING, scheduler._on_migrating), + mock.call(Event.ABORTING, scheduler._on_aborting), + ] -def test_close_stops_listening_for_the_migration(monkeypatch: pytest.MonkeyPatch, spider: DummySpider) -> None: - """`close` unregisters the migration listener, so a closed scheduler is not told about a migration.""" +def test_close_stops_listening_for_the_migration_and_the_abort( + monkeypatch: pytest.MonkeyPatch, + spider: DummySpider, +) -> None: + """`close` unregisters the listeners, so a closed scheduler is not told about a migration or an abort.""" actor = stub_scheduler_dependencies(monkeypatch) scheduler = ApifyScheduler(crawler=fake_crawler(scraper_busy=set())) scheduler.open(spider) scheduler.close('finished') - actor.off.assert_called_once_with(Event.MIGRATING, scheduler._on_migrating) + assert actor.off.call_args_list == [ + mock.call(Event.MIGRATING, scheduler._on_migrating), + mock.call(Event.ABORTING, scheduler._on_aborting), + ] def test_open_warns_when_the_actor_is_not_initialized( @@ -554,7 +563,7 @@ def test_open_warns_when_the_actor_is_not_initialized( spider: DummySpider, caplog: pytest.LogCaptureFixture, ) -> None: - """Without an initialized Actor there is nothing to register the migration listener with, and that is said.""" + """Without an initialized Actor there is nothing to register the listeners with, and that is said.""" actor = stub_scheduler_dependencies(monkeypatch) actor.on.side_effect = RuntimeError('The _ActorType is not active.') scheduler = ApifyScheduler(crawler=fake_crawler(scraper_busy=set())) @@ -600,6 +609,52 @@ async def test_migration_settles_the_requests_as_scrapy_finishes_them( assert rq.mark_request_as_handled.call_args_list == [mock.call(first_apify), mock.call(second_apify)] +async def test_abort_settles_the_requests_as_scrapy_finishes_them( + scheduler: ApifyScheduler, + monkeypatch: pytest.MonkeyPatch, + rq: mock.AsyncMock, +) -> None: + """Once an abort is announced, the requests Scrapy holds are marked as handled as they finish, not at the end.""" + monkeypatch.setattr('apify.scrapy.scheduler.SETTLE_POLL_INTERVAL', timedelta(milliseconds=10)) + busy = set(hand_out(scheduler, rq, 2)) + scheduler._crawler = fake_crawler(downloader_busy=busy) + (first_apify, first), (second_apify, second) = scheduler._requests_in_flight + + settled = asyncio.create_task(scheduler._on_aborting()) + await asyncio.sleep(0.05) + + rq.mark_request_as_handled.assert_not_called() + assert not settled.done() + + busy.discard(first) + await asyncio.sleep(0.05) + rq.mark_request_as_handled.assert_called_once_with(first_apify) + assert not settled.done() + + busy.discard(second) + await asyncio.wait_for(settled, timeout=1) + assert rq.mark_request_as_handled.call_args_list == [mock.call(first_apify), mock.call(second_apify)] + + +async def test_a_repeated_migration_announcement_does_not_settle_again( + scheduler: ApifyScheduler, + monkeypatch: pytest.MonkeyPatch, + rq: mock.AsyncMock, +) -> None: + """A second announcement, e.g. a reboot during a migration, returns at once while the first one keeps settling.""" + monkeypatch.setattr('apify.scrapy.scheduler.SETTLE_POLL_INTERVAL', timedelta(milliseconds=10)) + hand_out(scheduler, rq, 1) + + settled = asyncio.create_task(scheduler._on_migrating(EventMigratingData())) + await asyncio.sleep(0.05) + + await asyncio.wait_for(scheduler._on_migrating(EventMigratingData()), timeout=1) + assert not settled.done() + + scheduler.close('shutdown') + await asyncio.wait_for(settled, timeout=1) + + async def test_migration_with_nothing_in_flight_only_stops_handing_out_requests( scheduler: ApifyScheduler, rq: mock.AsyncMock, From 1952ce85a2f4503eb3f3f8d52507908795dfff2b Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 25 Aug 2026 19:32:40 +0200 Subject: [PATCH 5/9] refactor: unregister the Scrapy scheduler listeners without tracking whether they were registered --- src/apify/scrapy/extensions/_graceful_stop.py | 2 +- src/apify/scrapy/scheduler.py | 14 ++++---------- tests/unit/scrapy/test_scheduler.py | 4 ++-- 3 files changed, 7 insertions(+), 13 deletions(-) diff --git a/src/apify/scrapy/extensions/_graceful_stop.py b/src/apify/scrapy/extensions/_graceful_stop.py index 9e4c82c45..bb4339b0a 100644 --- a/src/apify/scrapy/extensions/_graceful_stop.py +++ b/src/apify/scrapy/extensions/_graceful_stop.py @@ -45,7 +45,7 @@ def spider_opened(self) -> None: def spider_closed(self) -> None: """Stop listening for the abort of the Actor run.""" - # The Actor may have exited already, in which case there is nothing left to unregister from. + # 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) diff --git a/src/apify/scrapy/scheduler.py b/src/apify/scrapy/scheduler.py index 9fb9b794d..3d3141d32 100644 --- a/src/apify/scrapy/scheduler.py +++ b/src/apify/scrapy/scheduler.py @@ -81,9 +81,6 @@ def __init__( self._migrating = False """Whether the platform announced a migration of the Actor run; nothing is handed out to Scrapy then.""" - self._listening = False - """Whether `_on_migrating` is registered with the Actor, so `close` knows to unregister it.""" - self._closed = False """Whether `close` has run; `_settle_requests_in_flight` stops then, as `close` resolves the rest itself.""" @@ -147,8 +144,6 @@ async def open_rq() -> RequestQueue: 'The Actor is not initialized, so the scheduler 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.' ) - else: - self._listening = True return None @@ -163,11 +158,10 @@ def close(self, reason: str) -> None: logger.debug(f'Closing {self.__class__.__name__} due to {reason}...') self._closed = True - if self._listening: - # The Actor may have exited already, in which case there is nothing left to unregister from. - with suppress(RuntimeError): - Actor.off(Event.MIGRATING, self._on_migrating) - Actor.off(Event.ABORTING, self._on_aborting) + # 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): diff --git a/tests/unit/scrapy/test_scheduler.py b/tests/unit/scrapy/test_scheduler.py index 3f1920c55..5c7f7cdc6 100644 --- a/tests/unit/scrapy/test_scheduler.py +++ b/tests/unit/scrapy/test_scheduler.py @@ -563,9 +563,10 @@ def test_open_warns_when_the_actor_is_not_initialized( spider: DummySpider, caplog: pytest.LogCaptureFixture, ) -> None: - """Without an initialized Actor there is nothing to register the listeners with, and that is said.""" + """Without an initialized Actor there is nothing to register the listeners with, and that is said; `close` copes.""" actor = stub_scheduler_dependencies(monkeypatch) actor.on.side_effect = RuntimeError('The _ActorType is not active.') + actor.off.side_effect = RuntimeError('The _ActorType is not active.') scheduler = ApifyScheduler(crawler=fake_crawler(scraper_busy=set())) with caplog.at_level(logging.WARNING, logger='apify.scrapy.scheduler'): @@ -573,7 +574,6 @@ def test_open_warns_when_the_actor_is_not_initialized( scheduler.close('finished') assert 'Actor is not initialized' in caplog.text - actor.off.assert_not_called() async def test_migration_settles_the_requests_as_scrapy_finishes_them( From 5fc47645849eceae01754320d86313d38a3470a5 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 25 Aug 2026 19:32:57 +0200 Subject: [PATCH 6/9] refactor: make the settle poll interval of the Scrapy scheduler private --- src/apify/scrapy/scheduler.py | 4 ++-- tests/unit/scrapy/test_scheduler.py | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/apify/scrapy/scheduler.py b/src/apify/scrapy/scheduler.py index 3d3141d32..ba8ad996d 100644 --- a/src/apify/scrapy/scheduler.py +++ b/src/apify/scrapy/scheduler.py @@ -33,7 +33,7 @@ logger = getLogger(__name__) -SETTLE_POLL_INTERVAL = timedelta(seconds=1) +_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.""" @@ -363,7 +363,7 @@ async def _settle_requests_in_flight(self) -> None: 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()) + 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. diff --git a/tests/unit/scrapy/test_scheduler.py b/tests/unit/scrapy/test_scheduler.py index 5c7f7cdc6..3316edaf9 100644 --- a/tests/unit/scrapy/test_scheduler.py +++ b/tests/unit/scrapy/test_scheduler.py @@ -582,7 +582,7 @@ async def test_migration_settles_the_requests_as_scrapy_finishes_them( rq: mock.AsyncMock, ) -> None: """Once a migration is announced, nothing more goes out and the requests Scrapy holds are marked as they finish.""" - monkeypatch.setattr('apify.scrapy.scheduler.SETTLE_POLL_INTERVAL', timedelta(milliseconds=10)) + monkeypatch.setattr('apify.scrapy.scheduler._SETTLE_POLL_INTERVAL', timedelta(milliseconds=10)) busy = set(hand_out(scheduler, rq, 2)) scheduler._crawler = fake_crawler(downloader_busy=busy) (first_apify, first), (second_apify, second) = scheduler._requests_in_flight @@ -615,7 +615,7 @@ async def test_abort_settles_the_requests_as_scrapy_finishes_them( rq: mock.AsyncMock, ) -> None: """Once an abort is announced, the requests Scrapy holds are marked as handled as they finish, not at the end.""" - monkeypatch.setattr('apify.scrapy.scheduler.SETTLE_POLL_INTERVAL', timedelta(milliseconds=10)) + monkeypatch.setattr('apify.scrapy.scheduler._SETTLE_POLL_INTERVAL', timedelta(milliseconds=10)) busy = set(hand_out(scheduler, rq, 2)) scheduler._crawler = fake_crawler(downloader_busy=busy) (first_apify, first), (second_apify, second) = scheduler._requests_in_flight @@ -642,7 +642,7 @@ async def test_a_repeated_migration_announcement_does_not_settle_again( rq: mock.AsyncMock, ) -> None: """A second announcement, e.g. a reboot during a migration, returns at once while the first one keeps settling.""" - monkeypatch.setattr('apify.scrapy.scheduler.SETTLE_POLL_INTERVAL', timedelta(milliseconds=10)) + monkeypatch.setattr('apify.scrapy.scheduler._SETTLE_POLL_INTERVAL', timedelta(milliseconds=10)) hand_out(scheduler, rq, 1) settled = asyncio.create_task(scheduler._on_migrating(EventMigratingData())) @@ -673,7 +673,7 @@ async def test_close_ends_the_migration_settling( rq: mock.AsyncMock, ) -> None: """Closing the scheduler while it settles a migration ends the settling; `close` resolves the rest itself.""" - monkeypatch.setattr('apify.scrapy.scheduler.SETTLE_POLL_INTERVAL', timedelta(milliseconds=10)) + monkeypatch.setattr('apify.scrapy.scheduler._SETTLE_POLL_INTERVAL', timedelta(milliseconds=10)) hand_out(scheduler, rq, 1) settled = asyncio.create_task(scheduler._on_migrating(EventMigratingData())) From eebd9c19500be520e0e1fe54aedcb673c2ccb03f Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 25 Aug 2026 19:33:19 +0200 Subject: [PATCH 7/9] test: stop leaking a worker thread per Scrapy scheduler test and tidy comments --- tests/e2e/test_actor_scrapy.py | 2 +- tests/unit/scrapy/test_scheduler.py | 4 ++-- tests/unit/scrapy/utils/test_apply_apify_settings.py | 1 + 3 files changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/e2e/test_actor_scrapy.py b/tests/e2e/test_actor_scrapy.py index 5c04c2c89..682d91402 100644 --- a/tests/e2e/test_actor_scrapy.py +++ b/tests/e2e/test_actor_scrapy.py @@ -42,7 +42,7 @@ async def test_actor_scrapy_title_spider( items = await actor.last_run().dataset().list_items() - # The start page and the pages it links to (`DEPTH_LIMIT` is 1 in the project settings), each scraped once. + # 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 diff --git a/tests/unit/scrapy/test_scheduler.py b/tests/unit/scrapy/test_scheduler.py index 3316edaf9..621050bea 100644 --- a/tests/unit/scrapy/test_scheduler.py +++ b/tests/unit/scrapy/test_scheduler.py @@ -63,10 +63,10 @@ def fake_async_thread(default_timeout: timedelta | None = None) -> mock.Mock: # never runs them would leave these tests asserting on the batching instead of on what reaches the RQ. Like the real thread, the coroutines run on a worker thread, so the double also works from within an async test. """ - executor = ThreadPoolExecutor(max_workers=1) def run_coro(coro: Coroutine, timeout: Any = 'default') -> Any: # noqa: ARG001 - return executor.submit(asyncio.run, coro).result() + with ThreadPoolExecutor(max_workers=1) as executor: + return executor.submit(asyncio.run, coro).result() def submit_coro(coro: Coroutine) -> Future: future: Future = Future() diff --git a/tests/unit/scrapy/utils/test_apply_apify_settings.py b/tests/unit/scrapy/utils/test_apply_apify_settings.py index c98f9164b..c13871ed1 100644 --- a/tests/unit/scrapy/utils/test_apply_apify_settings.py +++ b/tests/unit/scrapy/utils/test_apply_apify_settings.py @@ -49,6 +49,7 @@ def test_updates_downloader_middlewares() -> None: def test_registers_graceful_stop_extension() -> None: + """The graceful-stop extension is added to the extensions already configured, without displacing them.""" settings = Settings({'EXTENSIONS': {'scrapy.extensions.corestats.CoreStats': 500}}) new_settings = apply_apify_settings(settings=settings) From 8cc1e4ca1c17647652a9c25e0068dd927eb3f31a Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 25 Aug 2026 19:33:40 +0200 Subject: [PATCH 8/9] docs: mention the reboot timeout in the Scrapy migration section --- docs/03_guides/06_scrapy.mdx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/03_guides/06_scrapy.mdx b/docs/03_guides/06_scrapy.mdx index 08051915a..314a44a23 100644 --- a/docs/03_guides/06_scrapy.mdx +++ b/docs/03_guides/06_scrapy.mdx @@ -105,7 +105,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 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 whose callbacks are still running when the platform kills the process are downloaded again. +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. 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. From dcbc9a3044cbe4f6b2671dc9760ef29d0e1c5187 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 26 Aug 2026 12:41:42 +0200 Subject: [PATCH 9/9] refactor(scrapy): warn once per process when Scrapy runs without an initialized Actor --- src/apify/scrapy/_warnings.py | 22 ++++++++++++++++ src/apify/scrapy/extensions/_graceful_stop.py | 5 ++-- src/apify/scrapy/scheduler.py | 6 ++--- .../scrapy/extensions/test_graceful_stop.py | 9 +++++-- tests/unit/scrapy/test_scheduler.py | 4 ++- tests/unit/scrapy/test_warnings.py | 25 +++++++++++++++++++ 6 files changed, 61 insertions(+), 10 deletions(-) create mode 100644 src/apify/scrapy/_warnings.py create mode 100644 tests/unit/scrapy/test_warnings.py diff --git a/src/apify/scrapy/_warnings.py b/src/apify/scrapy/_warnings.py new file mode 100644 index 000000000..89a8ec494 --- /dev/null +++ b/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, + ) diff --git a/src/apify/scrapy/extensions/_graceful_stop.py b/src/apify/scrapy/extensions/_graceful_stop.py index bb4339b0a..30d147bdb 100644 --- a/src/apify/scrapy/extensions/_graceful_stop.py +++ b/src/apify/scrapy/extensions/_graceful_stop.py @@ -7,6 +7,7 @@ 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 @@ -39,9 +40,7 @@ def spider_opened(self) -> None: try: Actor.on(Event.ABORTING, self._on_aborting) except RuntimeError: - logger.warning( - 'The Actor is not initialized, so the crawl cannot be stopped gracefully when the run is aborted.' - ) + warn_about_uninitialized_actor() def spider_closed(self) -> None: """Stop listening for the abort of the Actor run.""" diff --git a/src/apify/scrapy/scheduler.py b/src/apify/scrapy/scheduler.py index ba8ad996d..9247da028 100644 --- a/src/apify/scrapy/scheduler.py +++ b/src/apify/scrapy/scheduler.py @@ -12,6 +12,7 @@ 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 Actor, Configuration, Event from apify.storage_clients import ApifyStorageClient @@ -140,10 +141,7 @@ async def open_rq() -> RequestQueue: Actor.on(Event.MIGRATING, self._on_migrating) Actor.on(Event.ABORTING, self._on_aborting) except RuntimeError: - logger.warning( - 'The Actor is not initialized, so the scheduler 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.' - ) + warn_about_uninitialized_actor() return None diff --git a/tests/unit/scrapy/extensions/test_graceful_stop.py b/tests/unit/scrapy/extensions/test_graceful_stop.py index ca7f26bcf..87dcf02f7 100644 --- a/tests/unit/scrapy/extensions/test_graceful_stop.py +++ b/tests/unit/scrapy/extensions/test_graceful_stop.py @@ -9,6 +9,7 @@ from scrapy.signalmanager import SignalManager from apify import Actor, Event, EventAbortingData +from apify.scrapy._warnings import logger_once from apify.scrapy.extensions import ApifyGracefulStopExtension if TYPE_CHECKING: @@ -62,11 +63,15 @@ async def test_the_actor_abort_event_reaches_the_extension() -> None: crawler.stop_async.assert_awaited_once_with() -def test_spider_opened_warns_when_the_actor_is_not_initialized(caplog: pytest.LogCaptureFixture) -> None: +def test_spider_opened_warns_when_the_actor_is_not_initialized( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: """Without an initialized Actor there is nothing to register the abort listener with, and that is said.""" + monkeypatch.setattr(logger_once, '_seen', set()) extension = ApifyGracefulStopExtension.from_crawler(fake_crawler()) - with caplog.at_level(logging.WARNING, logger='apify.scrapy.extensions._graceful_stop'): + with caplog.at_level(logging.WARNING, logger='apify.scrapy._warnings'): extension.spider_opened() extension.spider_closed() diff --git a/tests/unit/scrapy/test_scheduler.py b/tests/unit/scrapy/test_scheduler.py index 621050bea..fff15433d 100644 --- a/tests/unit/scrapy/test_scheduler.py +++ b/tests/unit/scrapy/test_scheduler.py @@ -16,6 +16,7 @@ from apify import Event, EventMigratingData from apify import Request as ApifyRequest from apify.scrapy._async_thread import AsyncThread +from apify.scrapy._warnings import logger_once from apify.scrapy.scheduler import ApifyScheduler from apify.storages import RequestQueue @@ -567,9 +568,10 @@ def test_open_warns_when_the_actor_is_not_initialized( actor = stub_scheduler_dependencies(monkeypatch) actor.on.side_effect = RuntimeError('The _ActorType is not active.') actor.off.side_effect = RuntimeError('The _ActorType is not active.') + monkeypatch.setattr(logger_once, '_seen', set()) scheduler = ApifyScheduler(crawler=fake_crawler(scraper_busy=set())) - with caplog.at_level(logging.WARNING, logger='apify.scrapy.scheduler'): + with caplog.at_level(logging.WARNING, logger='apify.scrapy._warnings'): scheduler.open(spider) scheduler.close('finished') diff --git a/tests/unit/scrapy/test_warnings.py b/tests/unit/scrapy/test_warnings.py new file mode 100644 index 000000000..c75b26a2f --- /dev/null +++ b/tests/unit/scrapy/test_warnings.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +from apify.scrapy._warnings import logger_once, warn_about_uninitialized_actor + +if TYPE_CHECKING: + import pytest + + +def test_the_uninitialized_actor_warning_is_logged_once( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """The scheduler and the extension both report the missing Actor, but the process logs the warning only once.""" + monkeypatch.setattr(logger_once, '_seen', set()) + + with caplog.at_level(logging.WARNING, logger='apify.scrapy._warnings'): + warn_about_uninitialized_actor() + warn_about_uninitialized_actor() + + warnings = [record for record in caplog.records if 'Actor is not initialized' in record.getMessage()] + assert len(warnings) == 1 + assert warnings[0].levelno == logging.WARNING