From bd79039b9a83ad8347b28a6d7521d320e980d32b Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 21 Aug 2026 13:36:29 +0200 Subject: [PATCH 1/5] fix(scrapy): stop silently dropping in-flight requests and redirects --- src/apify/scrapy/_async_thread.py | 28 +++++ src/apify/scrapy/requests.py | 30 ++--- src/apify/scrapy/scheduler.py | 106 ++++++++++++++---- .../scrapy/requests/test_to_apify_request.py | 24 ++++ tests/unit/scrapy/test_async_thread.py | 52 +++++++++ tests/unit/scrapy/test_scheduler.py | 90 ++++++++++++++- 6 files changed, 293 insertions(+), 37 deletions(-) diff --git a/src/apify/scrapy/_async_thread.py b/src/apify/scrapy/_async_thread.py index 90f6f4cb..6c543b27 100644 --- a/src/apify/scrapy/_async_thread.py +++ b/src/apify/scrapy/_async_thread.py @@ -74,6 +74,25 @@ def run_coro( future.cancel() raise + def submit_coro(self, coro: Coroutine) -> None: + """Schedule a coroutine on the event loop without waiting for its result. + + Use this for work whose result nothing depends on, so the calling thread is not blocked by the round + trip. Failures are logged, as there is no caller left to propagate them to, and a coroutine still + pending when `close` runs is cancelled along with the rest. + + Args: + coro: The coroutine to run. + + Raises: + RuntimeError: If the event loop has been closed. + """ + if self._eventloop.is_closed(): + raise RuntimeError(f'The coroutine {coro} cannot be executed because the event loop is closed.') + + future = asyncio.run_coroutine_threadsafe(coro, self._eventloop) + future.add_done_callback(self._log_failure) + def close(self, timeout: timedelta | None = None) -> None: """Close the event loop and its thread gracefully. @@ -110,6 +129,15 @@ def close(self, timeout: timedelta | None = None) -> None: logger.warning('Event loop thread did not exit cleanly! Forcing shutdown...') self._force_exit_event_loop() + @staticmethod + def _log_failure(future: futures.Future) -> None: + """Log the failure of a coroutine submitted via `submit_coro`.""" + if future.cancelled(): + return + + if (exc := future.exception()) is not None: + logger.error('A coroutine submitted to the event loop failed.', exc_info=exc) + def _start_event_loop(self) -> None: """Set up and run the asyncio event loop in the dedicated thread.""" asyncio.set_event_loop(self._eventloop) diff --git a/src/apify/scrapy/requests.py b/src/apify/scrapy/requests.py index 8bf99ec5..e8333886 100644 --- a/src/apify/scrapy/requests.py +++ b/src/apify/scrapy/requests.py @@ -76,8 +76,15 @@ def to_apify_request(scrapy_request: ScrapyRequest, spider: Spider) -> ApifyRequ try: if scrapy_request.dont_filter: request_kwargs['always_enqueue'] = True - elif scrapy_request.meta.get('apify_request_unique_key'): - request_kwargs['unique_key'] = scrapy_request.meta['apify_request_unique_key'] + # Reuse the queue's own unique key only while this is still the very request it was minted for. + # Scrapy derives new requests from a fetched one with `Request.replace()` (redirects) and spiders + # often forward `meta` verbatim to another URL; both inherit the stamp, and reusing it there would + # deduplicate the derived request against its parent and silently drop it. A stamp without a URL + # beside it was set by hand rather than by `to_scrapy_request`, so it is taken at face value. + elif (unique_key := scrapy_request.meta.get('apify_request_unique_key')) and ( + scrapy_request.meta.get('apify_request_url', scrapy_request.url) == scrapy_request.url + ): + request_kwargs['unique_key'] = unique_key # Serialize the Scrapy request now, before `Request.from_url()` runs below. `from_url()` mutates the # `user_data` dict it receives in place (it injects a live `CrawleeRequestData` under `__crawlee`), and that @@ -187,21 +194,14 @@ def to_scrapy_request(apify_request: ApifyRequest, spider: Spider) -> ScrapyRequ if not isinstance(scrapy_request, ScrapyRequest): raise TypeError('scrapy_request must be an instance of the ScrapyRequest class') - # Update the meta field with the meta field from the apify_request - meta = scrapy_request.meta or {} - meta.update({'apify_request_unique_key': apify_request.unique_key}) - # scrapy_request.meta is a property, so we have to set it like this - scrapy_request._meta = meta # noqa: SLF001 - # If the apify_request comes directly from the Scrapy, typically start URLs. else: - scrapy_request = ScrapyRequest( - url=apify_request.url, - method=apify_request.method, - meta={ - 'apify_request_unique_key': apify_request.unique_key, - }, - ) + scrapy_request = ScrapyRequest(url=apify_request.url, method=apify_request.method) + + # Stamp the queue's unique key together with the URL it belongs to, so that `to_apify_request` can tell + # this request apart from the ones Scrapy derives from it. + scrapy_request.meta['apify_request_unique_key'] = apify_request.unique_key + scrapy_request.meta['apify_request_url'] = scrapy_request.url # Add optional 'headers' field if apify_request.headers: diff --git a/src/apify/scrapy/scheduler.py b/src/apify/scrapy/scheduler.py index 0646d3d6..1a9f7924 100644 --- a/src/apify/scrapy/scheduler.py +++ b/src/apify/scrapy/scheduler.py @@ -19,6 +19,8 @@ from scrapy.http.request import Request from twisted.internet.defer import Deferred + from apify import Request as ApifyRequest + logger = getLogger(__name__) @@ -28,7 +30,11 @@ class ApifyScheduler(BaseScheduler): This scheduler requires the asyncio Twisted reactor to be installed. """ - def __init__(self, async_thread_timeout: timedelta = timedelta(seconds=60)) -> None: + def __init__( + self, + async_thread_timeout: timedelta = timedelta(seconds=60), + crawler: Crawler | None = None, + ) -> None: if not is_asyncio_reactor_installed(): raise ValueError( f'{ApifyScheduler.__qualname__} requires the asyncio Twisted reactor. ' @@ -37,6 +43,10 @@ def __init__(self, async_thread_timeout: timedelta = timedelta(seconds=60)) -> N ) self._rq: RequestQueue | None = None self.spider: Spider | None = None + self._crawler = crawler + + self._requests_in_flight: list[tuple[ApifyRequest, Request]] = [] + """Requests handed over to Scrapy and not resolved in the request queue yet.""" # A thread with the asyncio event loop to run coroutines on. self._async_thread = AsyncThread(default_timeout=async_thread_timeout) @@ -49,7 +59,7 @@ def from_crawler(cls, crawler: Crawler) -> ApifyScheduler: background event loop may take before timing out; it defaults to 60 seconds. """ timeout_secs = crawler.settings.getint('APIFY_ASYNC_THREAD_TIMEOUT_SECS', 60) - return cls(async_thread_timeout=timedelta(seconds=timeout_secs)) + return cls(async_thread_timeout=timedelta(seconds=timeout_secs), crawler=crawler) def open(self, spider: Spider) -> Deferred[None] | None: """Open the scheduler. @@ -86,12 +96,26 @@ async def open_rq() -> RequestQueue: def close(self, reason: str) -> None: """Close the scheduler. - Shut down the event loop and its thread gracefully. + Resolve the requests Scrapy still holds, then shut down the event loop and its thread gracefully. Args: reason: The reason for closing the spider. """ logger.debug(f'Closing {self.__class__.__name__} due to {reason}...') + + rq = self._rq + if isinstance(rq, RequestQueue): + try: + # Resolve what Scrapy holds while the event loop is still around. Whatever it did not finish - + # an interrupted run, an Actor migration - goes back to the queue, so the next run picks it up + # instead of waiting for its lock to expire. + self._resolve_finished_requests(wait=True) + for apify_request, _ in self._requests_in_flight: + self._async_thread.run_coro(rq.reclaim_request(apify_request)) + self._requests_in_flight.clear() + except Exception: + logger.exception('Failed to resolve the requests still in flight in the request queue.') + try: self._async_thread.close() @@ -113,6 +137,11 @@ def has_pending_requests(self) -> bool: if not isinstance(self._rq, RequestQueue): raise TypeError('self._rq must be an instance of the RequestQueue class') + # Scrapy asks this only once both its downloader and its scraper are idle, so everything still tracked + # as in flight is provably finished. Wait for those updates to land: the queue reports itself unfinished + # while any request it handed out is still unresolved. + self._resolve_finished_requests(wait=True) + # Log here before re-raising: this coroutine ran on a separate event-loop thread, and the failure is # otherwise easy to lose as it crosses that thread boundary back into Scrapy's synchronous machinery. try: @@ -164,6 +193,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') + # Resolve whatever Scrapy has finished since the last call. The engine polls this method throughout the + # crawl, which keeps the queue's view of progress current without blocking on the round trips. + self._resolve_finished_requests(wait=False) + # Log here before re-raising: this coroutine ran on a separate event-loop thread, and the failure is # otherwise easy to lose as it crosses that thread boundary back into Scrapy's synchronous machinery. try: @@ -178,26 +211,61 @@ def next_request(self) -> Request | None: if not isinstance(self.spider, Spider): raise TypeError('self.spider must be an instance of the Spider class') - # Reconstruct the Scrapy request before consuming the queue entry. A malformed entry must not crash - # the whole run, so on failure it is logged and skipped (None) rather than propagating. + # A malformed entry must not crash the whole run, so on failure it is logged and skipped rather than + # propagating. Such an unrecoverable entry (a corrupt or legacy payload) is marked as handled right + # away, otherwise the queue would keep handing it back forever. try: scrapy_request = to_scrapy_request(apify_request, spider=self.spider) except Exception as exc: logger.warning(f'Failed to convert Apify request {apify_request} to a Scrapy request; skipping it: {exc}') - scrapy_request = None - - # Mark the request as handled. This runs even when reconstruction failed above: an unrecoverable entry - # (a corrupt or legacy payload) must still be consumed, otherwise the queue would keep handing it back - # forever. Retrying genuine failures is the RetryMiddleware's job. - # Log here before re-raising: this coroutine ran on a separate event-loop thread, and the failure is - # otherwise easy to lose as it crosses that thread boundary back into Scrapy's synchronous machinery. - try: - self._async_thread.run_coro(self._rq.mark_request_as_handled(apify_request)) - except Exception: - logger.exception('Failed to mark the request as handled in the request queue.') - raise - - if scrapy_request is None: + try: + self._async_thread.run_coro(self._rq.mark_request_as_handled(apify_request)) + except Exception: + logger.exception('Failed to mark the request as handled in the request queue.') + raise return None + # The entry stays unresolved in the queue until Scrapy is done with the request, so a run interrupted + # mid-flight leaves it pending instead of silently handled. + self._requests_in_flight.append((apify_request, scrapy_request)) + return scrapy_request + + def _requests_busy_in_scrapy(self) -> set[Request]: + """Return the requests Scrapy is still working on. + + A request handed out by `next_request` stays in the downloader's or the scraper's active set until its + download, the downloader middleware chain and the spider callback have all finished, so absence from + both means Scrapy is done with it. Requests a middleware drops before the download reach neither set, + but Scrapy only asks `has_pending_requests` once both are empty, which is what settles those too. + """ + engine = self._crawler.engine if self._crawler is not None else None + if engine is None: + return set() + + scraper_slot = engine.scraper.slot + return engine.downloader.active | (scraper_slot.active if scraper_slot is not None else set()) + + def _resolve_finished_requests(self, *, wait: bool) -> None: + """Mark every request Scrapy has finished processing as handled in the request queue. + + Args: + wait: Whether to block until the queue has been updated. Pass False on the crawl's hot path, where + nothing depends on the result and blocking would stall the Twisted reactor. + """ + rq = self._rq + if not self._requests_in_flight or not isinstance(rq, RequestQueue): + return + + busy = self._requests_busy_in_scrapy() + still_in_flight = [] + + for apify_request, scrapy_request in self._requests_in_flight: + if scrapy_request in busy: + still_in_flight.append((apify_request, scrapy_request)) + elif wait: + self._async_thread.run_coro(rq.mark_request_as_handled(apify_request)) + else: + self._async_thread.submit_coro(rq.mark_request_as_handled(apify_request)) + + self._requests_in_flight = still_in_flight diff --git a/tests/unit/scrapy/requests/test_to_apify_request.py b/tests/unit/scrapy/requests/test_to_apify_request.py index 97902f7d..1ce15fe9 100644 --- a/tests/unit/scrapy/requests/test_to_apify_request.py +++ b/tests/unit/scrapy/requests/test_to_apify_request.py @@ -10,6 +10,7 @@ from crawlee._types import HttpHeaders +from apify import Request as ApifyRequest from apify.scrapy.requests import to_apify_request, to_scrapy_request @@ -187,3 +188,26 @@ def test_apify_request_id_in_meta_is_ignored(spider: Spider) -> None: assert apify_request is not None assert apify_request.unique_key == 'https://example.com' + + +def test_redirected_request_does_not_inherit_the_parents_unique_key(spider: Spider) -> None: + """A redirect derived from a fetched request gets its own unique key instead of the parent's stamp.""" + parent = to_scrapy_request(ApifyRequest.from_url('https://example.com/redirect'), spider) + redirected = parent.replace(url='https://example.com/target') + + apify_request = to_apify_request(redirected, spider) + + assert apify_request is not None + assert apify_request.url == 'https://example.com/target' + assert apify_request.unique_key != parent.meta['apify_request_unique_key'] + + +def test_follow_up_request_with_propagated_meta_gets_its_own_unique_key(spider: Spider) -> None: + """A spider callback forwarding `meta` verbatim to another URL must not reuse the parent's unique key.""" + parent = to_scrapy_request(ApifyRequest.from_url('https://example.com/listing'), spider) + follow_up = Request(url='https://example.com/detail', meta=parent.meta) + + apify_request = to_apify_request(follow_up, spider) + + assert apify_request is not None + assert apify_request.unique_key != parent.meta['apify_request_unique_key'] diff --git a/tests/unit/scrapy/test_async_thread.py b/tests/unit/scrapy/test_async_thread.py index 3cf51b62..df0f52c2 100644 --- a/tests/unit/scrapy/test_async_thread.py +++ b/tests/unit/scrapy/test_async_thread.py @@ -161,3 +161,55 @@ async def boom() -> None: # The loop was stopped and its thread joined despite the failing cancellation, so nothing is left running. assert not thread._thread.is_alive() assert thread._eventloop.is_closed() + + +def test_submit_coro_runs_the_coroutine_without_blocking() -> None: + """`submit_coro` schedules the coroutine on the background loop and returns before it completes.""" + thread = AsyncThread() + _wait_until_running(thread) + + release = threading.Event() + finished = threading.Event() + + async def gated() -> None: + await asyncio.to_thread(release.wait) + finished.set() + + thread.submit_coro(gated()) + + # The call returned while the coroutine is still parked on the gate. + assert not finished.is_set() + + release.set() + assert finished.wait(timeout=2) + + thread.close() + + +def test_submit_coro_logs_a_failing_coroutine(caplog: pytest.LogCaptureFixture) -> None: + """A coroutine submitted without a caller to propagate to has its failure logged instead of swallowed.""" + thread = AsyncThread() + _wait_until_running(thread) + + async def boom() -> None: + raise RuntimeError('boom') + + with caplog.at_level(logging.ERROR, logger='apify.scrapy._async_thread'): + thread.submit_coro(boom()) + thread.close() + + errors = [record for record in caplog.records if record.levelno >= logging.ERROR] + assert len(errors) == 1 + assert errors[0].exc_info is not None + assert str(errors[0].exc_info[1]) == 'boom' + + +def test_submit_coro_raises_after_close() -> None: + """`submit_coro` raises `RuntimeError` once the loop has been closed.""" + thread = AsyncThread() + thread.close() + + coro = _return(42) + with pytest.raises(RuntimeError): + thread.submit_coro(coro) + coro.close() diff --git a/tests/unit/scrapy/test_scheduler.py b/tests/unit/scrapy/test_scheduler.py index a7cc4445..5413a3ac 100644 --- a/tests/unit/scrapy/test_scheduler.py +++ b/tests/unit/scrapy/test_scheduler.py @@ -25,6 +25,15 @@ def spider() -> DummySpider: return DummySpider() +def fake_crawler(busy: set[Request]) -> Any: + """Build a crawler stub whose engine reports `busy` as the requests Scrapy is still working on.""" + engine = SimpleNamespace( + downloader=SimpleNamespace(active=busy), + scraper=SimpleNamespace(slot=None), + ) + return SimpleNamespace(engine=engine) + + @pytest.fixture def scheduler(monkeypatch: pytest.MonkeyPatch, spider: DummySpider) -> ApifyScheduler: """Create a scheduler with its reactor check and async thread stubbed out.""" @@ -124,8 +133,7 @@ def test_next_request_skips_request_that_fails_to_convert( def test_next_request_returns_converted_request(scheduler: ApifyScheduler) -> None: - """A valid queue entry is reconstructed into a Scrapy request and marked handled.""" - rq = cast('mock.MagicMock', scheduler._rq) + """A valid queue entry is reconstructed into a Scrapy request.""" async_thread = cast('mock.MagicMock', scheduler._async_thread) apify_request = ApifyRequest( @@ -140,7 +148,6 @@ def test_next_request_returns_converted_request(scheduler: ApifyScheduler) -> No assert isinstance(result, Request) assert result.url == apify_request.url - rq.mark_request_as_handled.assert_called_once_with(apify_request) def test_next_request_returns_none_when_queue_empty(scheduler: ApifyScheduler) -> None: @@ -190,3 +197,80 @@ def __init__(self, default_timeout: timedelta | None = None) -> None: ApifyScheduler.from_crawler(cast('Any', crawler)) assert captured['default_timeout'] == timedelta(seconds=123) + + +def test_next_request_leaves_the_request_unhandled(scheduler: ApifyScheduler) -> None: + """A request handed to Scrapy stays unhandled in the queue until Scrapy has finished processing it.""" + rq = cast('mock.MagicMock', scheduler._rq) + async_thread = cast('mock.MagicMock', scheduler._async_thread) + + apify_request = ApifyRequest( + url='https://example.com', + method='GET', + unique_key='https://example.com', + user_data={}, + ) + async_thread.run_coro.return_value = apify_request + + result = scheduler.next_request() + + assert isinstance(result, Request) + rq.mark_request_as_handled.assert_not_called() + + +def test_has_pending_requests_marks_finished_requests_as_handled(scheduler: ApifyScheduler) -> None: + """Requests Scrapy has finished with are marked as handled once it goes idle and asks about pending work.""" + rq = cast('mock.MagicMock', scheduler._rq) + async_thread = cast('mock.MagicMock', scheduler._async_thread) + + apify_request = ApifyRequest( + url='https://example.com', + method='GET', + unique_key='https://example.com', + user_data={}, + ) + async_thread.run_coro.return_value = apify_request + scheduler.next_request() + rq.mark_request_as_handled.assert_not_called() + + # Scrapy asks about pending work only once it is idle, which is also how a request a middleware dropped + # before the download - an offsite or robots.txt denial - is settled. + scheduler._crawler = fake_crawler(busy=set()) + async_thread.run_coro.return_value = True # the queue reports itself finished + assert scheduler.has_pending_requests() is False + + rq.mark_request_as_handled.assert_called_once_with(apify_request) + + +def test_close_reclaims_requests_scrapy_never_finished(scheduler: ApifyScheduler) -> None: + """Requests still being processed when the scheduler closes go back to the queue instead of being lost.""" + rq = cast('mock.MagicMock', scheduler._rq) + async_thread = cast('mock.MagicMock', scheduler._async_thread) + + apify_request = ApifyRequest( + url='https://example.com', + method='GET', + unique_key='https://example.com', + user_data={}, + ) + async_thread.run_coro.return_value = apify_request + scrapy_request = scheduler.next_request() + + # Scrapy is still downloading the request when the run is interrupted. + scheduler._crawler = fake_crawler(busy={cast('Request', scrapy_request)}) + + scheduler.close('shutdown') + + rq.reclaim_request.assert_called_once_with(apify_request) + rq.mark_request_as_handled.assert_not_called() + + +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.""" + monkeypatch.setattr('apify.scrapy.scheduler.is_asyncio_reactor_installed', lambda: True) + monkeypatch.setattr('apify.scrapy.scheduler.AsyncThread', mock.MagicMock()) + + crawler = SimpleNamespace(settings=Settings()) + scheduler = ApifyScheduler.from_crawler(cast('Any', crawler)) + + assert scheduler._crawler is crawler From 08251656e1d86212c18bb6cb27317ed2d6b5e343 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 21 Aug 2026 18:50:03 +0200 Subject: [PATCH 2/5] fix(scrapy): resolve request queue updates reliably on the shutdown and failure paths --- src/apify/scrapy/_async_thread.py | 35 ++++- src/apify/scrapy/scheduler.py | 63 ++++++--- tests/unit/scrapy/test_async_thread.py | 64 ++++++++- tests/unit/scrapy/test_scheduler.py | 182 +++++++++++++++++++++---- 4 files changed, 300 insertions(+), 44 deletions(-) diff --git a/src/apify/scrapy/_async_thread.py b/src/apify/scrapy/_async_thread.py index 6c543b27..3e3d57ee 100644 --- a/src/apify/scrapy/_async_thread.py +++ b/src/apify/scrapy/_async_thread.py @@ -12,6 +12,9 @@ logger = getLogger(__name__) +_SUBMITTED_PRUNE_THRESHOLD = 128 +"""How many `submit_coro` futures may pile up before the finished ones are dropped from the tracking list.""" + class AsyncThread: """Run an asyncio event loop in a dedicated background thread. @@ -26,6 +29,9 @@ def __init__(self, default_timeout: timedelta = timedelta(seconds=60)) -> None: self._default_timeout = default_timeout self._eventloop = asyncio.new_event_loop() + self._submitted: list[futures.Future] = [] + """Futures of the coroutines submitted via `submit_coro` that may still be running.""" + # Start the event loop in a dedicated daemon thread. self._thread = threading.Thread( target=self._start_event_loop, @@ -79,7 +85,8 @@ def submit_coro(self, coro: Coroutine) -> None: Use this for work whose result nothing depends on, so the calling thread is not blocked by the round trip. Failures are logged, as there is no caller left to propagate them to, and a coroutine still - pending when `close` runs is cancelled along with the rest. + pending when `close` runs is cancelled along with the rest - call `wait_for_submitted` before anything + that must not see that happen. Args: coro: The coroutine to run. @@ -90,8 +97,34 @@ def submit_coro(self, coro: Coroutine) -> None: if self._eventloop.is_closed(): raise RuntimeError(f'The coroutine {coro} cannot be executed because the event loop is closed.') + # Drop the futures that already finished. `wait_for_submitted` only runs once Scrapy goes idle, so + # without this the list would hold every coroutine the whole crawl ever submitted, with its result. + if len(self._submitted) >= _SUBMITTED_PRUNE_THRESHOLD: + self._submitted = [submitted for submitted in self._submitted if not submitted.done()] + future = asyncio.run_coroutine_threadsafe(coro, self._eventloop) future.add_done_callback(self._log_failure) + self._submitted.append(future) + + def wait_for_submitted(self, timeout: timedelta | None = None) -> None: + """Block until the coroutines submitted via `submit_coro` have finished. + + Use this before anything that would observe their effects, or before `close`, which cancels whatever is + still running. Coroutines that do not finish within the timeout stay tracked for the next call. + + Args: + timeout: The maximum time to wait for the submitted coroutines. Pass `None` to use the + `default_timeout` passed to the constructor. + """ + if timeout is None: + timeout = self._default_timeout + + self._submitted = list(futures.wait(self._submitted, timeout=timeout.total_seconds()).not_done) + + # Returning with coroutines still pending breaks the guarantee the callers rely on, so say so rather + # than letting them act on effects that have not landed. + if self._submitted: + logger.warning(f'{len(self._submitted)} submitted coroutines did not finish within the timeout.') def close(self, timeout: timedelta | None = None) -> None: """Close the event loop and its thread gracefully. diff --git a/src/apify/scrapy/scheduler.py b/src/apify/scrapy/scheduler.py index 1a9f7924..94096176 100644 --- a/src/apify/scrapy/scheduler.py +++ b/src/apify/scrapy/scheduler.py @@ -106,16 +106,25 @@ def close(self, reason: str) -> None: rq = self._rq if isinstance(rq, RequestQueue): try: - # Resolve what Scrapy holds while the event loop is still around. Whatever it did not finish - - # an interrupted run, an Actor migration - goes back to the queue, so the next run picks it up - # instead of waiting for its lock to expire. self._resolve_finished_requests(wait=True) - for apify_request, _ in self._requests_in_flight: - self._async_thread.run_coro(rq.reclaim_request(apify_request)) - self._requests_in_flight.clear() except Exception: logger.exception('Failed to resolve the requests still in flight in the request queue.') + # Whatever Scrapy did not finish - an interrupted run, an Actor migration - goes back to the queue + # while the event loop is still around, so the next run gets it as pending. Each request is + # reclaimed on its own, so one failure does not strand the rest. + for apify_request, _ in self._requests_in_flight: + try: + self._async_thread.run_coro(rq.reclaim_request(apify_request)) + except Exception: + logger.exception(f'Failed to reclaim the request {apify_request} in the request queue.') + + self._requests_in_flight.clear() + + # Let the updates fired off on the hot path finish: closing the event loop cancels them silently, which + # would leave those requests unhandled in the queue. + self._async_thread.wait_for_submitted() + try: self._async_thread.close() @@ -131,6 +140,8 @@ def close(self, reason: str) -> None: def has_pending_requests(self) -> bool: """Check if the scheduler has any pending requests. + Resolves the requests Scrapy has finished with first, as their outcome is what decides the answer. + Returns: True if the scheduler has any pending requests, False otherwise. """ @@ -138,10 +149,13 @@ def has_pending_requests(self) -> bool: raise TypeError('self._rq must be an instance of the RequestQueue class') # Scrapy asks this only once both its downloader and its scraper are idle, so everything still tracked - # as in flight is provably finished. Wait for those updates to land: the queue reports itself unfinished - # while any request it handed out is still unresolved. + # as in flight is provably finished. self._resolve_finished_requests(wait=True) + # The queue answers from its own bookkeeping, so an update still in flight would let it report itself + # finished while a request is unhandled - and closing the crawl would then cancel that update. + self._async_thread.wait_for_submitted() + # Log here before re-raising: this coroutine ran on a separate event-loop thread, and the failure is # otherwise easy to lose as it crosses that thread boundary back into Scrapy's synchronous machinery. try: @@ -234,10 +248,10 @@ def next_request(self) -> Request | None: def _requests_busy_in_scrapy(self) -> set[Request]: """Return the requests Scrapy is still working on. - A request handed out by `next_request` stays in the downloader's or the scraper's active set until its - download, the downloader middleware chain and the spider callback have all finished, so absence from - both means Scrapy is done with it. Requests a middleware drops before the download reach neither set, - but Scrapy only asks `has_pending_requests` once both are empty, which is what settles those too. + A request handed out by `next_request` joins the downloader's active set before the middleware chain + runs, and leaves the scraper's only once the spider callback and the item pipeline have finished with + it. Absence from both therefore means Scrapy is done with the request, whether it was downloaded, + dropped by a middleware or errored out. """ engine = self._crawler.engine if self._crawler is not None else None if engine is None: @@ -249,6 +263,11 @@ def _requests_busy_in_scrapy(self) -> set[Request]: def _resolve_finished_requests(self, *, wait: bool) -> None: """Mark every request Scrapy has finished processing as handled in the request queue. + Only a failure to dispatch the update - a timed-out or closed event loop - keeps a request tracked for + the next call to retry, without stopping the rest of the list from being resolved. The queue reports + the update's own failures by returning `None`, which is indistinguishable from success here, so such a + request is left unhandled for the next run to pick up. + Args: wait: Whether to block until the queue has been updated. Pass False on the crawl's hot path, where nothing depends on the result and blocking would stall the Twisted reactor. @@ -258,14 +277,20 @@ def _resolve_finished_requests(self, *, wait: bool) -> None: return busy = self._requests_busy_in_scrapy() - still_in_flight = [] + unresolved: list[tuple[ApifyRequest, Request]] = [] for apify_request, scrapy_request in self._requests_in_flight: if scrapy_request in busy: - still_in_flight.append((apify_request, scrapy_request)) - elif wait: - self._async_thread.run_coro(rq.mark_request_as_handled(apify_request)) - else: - self._async_thread.submit_coro(rq.mark_request_as_handled(apify_request)) + unresolved.append((apify_request, scrapy_request)) + continue + + try: + if wait: + self._async_thread.run_coro(rq.mark_request_as_handled(apify_request)) + else: + self._async_thread.submit_coro(rq.mark_request_as_handled(apify_request)) + except Exception: + logger.exception(f'Failed to mark the request {apify_request} as handled in the request queue.') + unresolved.append((apify_request, scrapy_request)) - self._requests_in_flight = still_in_flight + self._requests_in_flight = unresolved diff --git a/tests/unit/scrapy/test_async_thread.py b/tests/unit/scrapy/test_async_thread.py index df0f52c2..994db6e8 100644 --- a/tests/unit/scrapy/test_async_thread.py +++ b/tests/unit/scrapy/test_async_thread.py @@ -11,7 +11,7 @@ import pytest from ..._utils import poll_until_condition -from apify.scrapy._async_thread import AsyncThread +from apify.scrapy._async_thread import _SUBMITTED_PRUNE_THRESHOLD, AsyncThread async def _return(value: int) -> int: @@ -213,3 +213,65 @@ def test_submit_coro_raises_after_close() -> None: with pytest.raises(RuntimeError): thread.submit_coro(coro) coro.close() + + +def test_wait_for_submitted_blocks_until_the_coroutines_finish() -> None: + """`wait_for_submitted` waits for the fire-and-forget coroutines, so `close` cannot cancel them.""" + thread = AsyncThread() + _wait_until_running(thread) + + release = threading.Event() + finished = threading.Event() + + async def gated() -> None: + await asyncio.to_thread(release.wait) + finished.set() + + thread.submit_coro(gated()) + release.set() + + thread.wait_for_submitted() + + assert finished.is_set() + thread.close() + + +def test_wait_for_submitted_keeps_an_unfinished_coroutine_tracked(caplog: pytest.LogCaptureFixture) -> None: + """A coroutine that outlasts the timeout stays tracked and is reported, so a later call can wait for it.""" + thread = AsyncThread() + _wait_until_running(thread) + + release = threading.Event() + + async def gated() -> None: + await asyncio.to_thread(release.wait) + + thread.submit_coro(gated()) + + with caplog.at_level(logging.WARNING, logger='apify.scrapy._async_thread'): + thread.wait_for_submitted(timeout=timedelta(seconds=0.01)) + assert len(thread._submitted) == 1 + assert [record for record in caplog.records if record.levelno == logging.WARNING] + + release.set() + thread.wait_for_submitted() + assert thread._submitted == [] + + thread.close() + + +def test_submit_coro_drops_the_finished_futures() -> None: + """Only the coroutines still running stay tracked, so a long crawl does not pile up finished futures.""" + thread = AsyncThread() + _wait_until_running(thread) + + for _ in range(_SUBMITTED_PRUNE_THRESHOLD): + thread.submit_coro(_return(1)) + + assert futures.wait(list(thread._submitted), timeout=2).not_done == set() + + # Every tracked coroutine has finished, so this submission drops them instead of growing the list. + thread.submit_coro(_return(1)) + assert len(thread._submitted) == 1 + + thread.close() diff --git a/tests/unit/scrapy/test_scheduler.py b/tests/unit/scrapy/test_scheduler.py index 5413a3ac..c4032955 100644 --- a/tests/unit/scrapy/test_scheduler.py +++ b/tests/unit/scrapy/test_scheduler.py @@ -25,11 +25,16 @@ def spider() -> DummySpider: return DummySpider() -def fake_crawler(busy: set[Request]) -> Any: - """Build a crawler stub whose engine reports `busy` as the requests Scrapy is still working on.""" +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.""" + scraper_slot = SimpleNamespace(active=scraper_busy) if scraper_busy is not None else None engine = SimpleNamespace( - downloader=SimpleNamespace(active=busy), - scraper=SimpleNamespace(slot=None), + downloader=SimpleNamespace(active=downloader_busy if downloader_busy is not None else set()), + scraper=SimpleNamespace(slot=scraper_slot), ) return SimpleNamespace(engine=engine) @@ -133,7 +138,8 @@ def test_next_request_skips_request_that_fails_to_convert( def test_next_request_returns_converted_request(scheduler: ApifyScheduler) -> None: - """A valid queue entry is reconstructed into a Scrapy request.""" + """A valid queue entry is reconstructed into a Scrapy request and left unhandled until Scrapy is done.""" + rq = cast('mock.MagicMock', scheduler._rq) async_thread = cast('mock.MagicMock', scheduler._async_thread) apify_request = ApifyRequest( @@ -142,12 +148,13 @@ def test_next_request_returns_converted_request(scheduler: ApifyScheduler) -> No unique_key='https://example.com', user_data={}, ) - async_thread.run_coro.side_effect = [apify_request, None] + async_thread.run_coro.return_value = apify_request result = scheduler.next_request() assert isinstance(result, Request) assert result.url == apify_request.url + rq.mark_request_as_handled.assert_not_called() def test_next_request_returns_none_when_queue_empty(scheduler: ApifyScheduler) -> None: @@ -199,8 +206,8 @@ def __init__(self, default_timeout: timedelta | None = None) -> None: assert captured['default_timeout'] == timedelta(seconds=123) -def test_next_request_leaves_the_request_unhandled(scheduler: ApifyScheduler) -> None: - """A request handed to Scrapy stays unhandled in the queue until Scrapy has finished processing it.""" +def test_has_pending_requests_marks_finished_requests_as_handled(scheduler: ApifyScheduler) -> None: + """Requests Scrapy has finished with are marked as handled once it goes idle and asks about pending work.""" rq = cast('mock.MagicMock', scheduler._rq) async_thread = cast('mock.MagicMock', scheduler._async_thread) @@ -211,15 +218,19 @@ def test_next_request_leaves_the_request_unhandled(scheduler: ApifyScheduler) -> user_data={}, ) async_thread.run_coro.return_value = apify_request + scheduler.next_request() + rq.mark_request_as_handled.assert_not_called() - result = scheduler.next_request() + # Scrapy asks about pending work only once its downloader and its scraper are both idle. + scheduler._crawler = fake_crawler(scraper_busy=set()) + async_thread.run_coro.return_value = True # the queue reports itself finished + assert scheduler.has_pending_requests() is False - assert isinstance(result, Request) - rq.mark_request_as_handled.assert_not_called() + rq.mark_request_as_handled.assert_called_once_with(apify_request) -def test_has_pending_requests_marks_finished_requests_as_handled(scheduler: ApifyScheduler) -> None: - """Requests Scrapy has finished with are marked as handled once it goes idle and asks about pending work.""" +def test_next_request_marks_finished_requests_without_blocking(scheduler: ApifyScheduler) -> None: + """On the crawl's hot path a finished request is marked as handled without blocking the reactor on it.""" rq = cast('mock.MagicMock', scheduler._rq) async_thread = cast('mock.MagicMock', scheduler._async_thread) @@ -230,19 +241,47 @@ def test_has_pending_requests_marks_finished_requests_as_handled(scheduler: Apif user_data={}, ) async_thread.run_coro.return_value = apify_request - scheduler.next_request() - rq.mark_request_as_handled.assert_not_called() + scrapy_request = scheduler.next_request() - # Scrapy asks about pending work only once it is idle, which is also how a request a middleware dropped - # before the download - an offsite or robots.txt denial - is settled. - scheduler._crawler = fake_crawler(busy=set()) - async_thread.run_coro.return_value = True # the queue reports itself finished - assert scheduler.has_pending_requests() is False + # The queue is drained from here on, so no further request is handed out. + async_thread.run_coro.return_value = None + + # Scrapy is still downloading the request, so it stays unresolved. + scheduler._crawler = fake_crawler(downloader_busy={cast('Request', scrapy_request)}) + assert scheduler.next_request() is None + async_thread.submit_coro.assert_not_called() + + # Scrapy is done with it, so it is resolved off the reactor thread instead of blocking on the round trip. + scheduler._crawler = fake_crawler(scraper_busy=set()) + assert scheduler.next_request() is None rq.mark_request_as_handled.assert_called_once_with(apify_request) + async_thread.submit_coro.assert_called_once_with(rq.mark_request_as_handled.return_value) + + +def test_has_pending_requests_waits_for_the_non_blocking_updates(scheduler: ApifyScheduler) -> None: + """The queue is asked whether it is finished only after the updates fired off on the hot path have landed.""" + async_thread = cast('mock.MagicMock', scheduler._async_thread) + scheduler._crawler = fake_crawler() + async_thread.run_coro.return_value = True # the queue reports itself finished + assert scheduler.has_pending_requests() is False -def test_close_reclaims_requests_scrapy_never_finished(scheduler: ApifyScheduler) -> None: + # The queue answers from its own bookkeeping, which a pending update has not reached yet. + assert async_thread.mock_calls == [ + mock.call.wait_for_submitted(), + mock.call.run_coro(cast('mock.MagicMock', scheduler._rq).is_finished()), + ] + + +@pytest.mark.parametrize( + 'busy_kwarg', + [ + pytest.param('downloader_busy', id='busy in the downloader'), + pytest.param('scraper_busy', id='busy in the scraper slot'), + ], +) +def test_close_reclaims_requests_scrapy_never_finished(scheduler: ApifyScheduler, busy_kwarg: str) -> None: """Requests still being processed when the scheduler closes go back to the queue instead of being lost.""" rq = cast('mock.MagicMock', scheduler._rq) async_thread = cast('mock.MagicMock', scheduler._async_thread) @@ -256,8 +295,8 @@ def test_close_reclaims_requests_scrapy_never_finished(scheduler: ApifyScheduler async_thread.run_coro.return_value = apify_request scrapy_request = scheduler.next_request() - # Scrapy is still downloading the request when the run is interrupted. - scheduler._crawler = fake_crawler(busy={cast('Request', scrapy_request)}) + # Scrapy is still working on the request when the run is interrupted. + scheduler._crawler = fake_crawler(**{busy_kwarg: {cast('Request', scrapy_request)}}) scheduler.close('shutdown') @@ -265,6 +304,103 @@ 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: + """Requests Scrapy drained before the shutdown are marked as handled rather than reclaimed.""" + rq = cast('mock.MagicMock', scheduler._rq) + async_thread = cast('mock.MagicMock', scheduler._async_thread) + + apify_request = ApifyRequest( + url='https://example.com', + method='GET', + unique_key='https://example.com', + user_data={}, + ) + async_thread.run_coro.return_value = apify_request + scheduler.next_request() + + # Scrapy drains its downloader and its scraper before the scheduler is closed. + scheduler._crawler = fake_crawler(scraper_busy=set()) + + scheduler.close('finished') + + rq.mark_request_as_handled.assert_called_once_with(apify_request) + rq.reclaim_request.assert_not_called() + + +def test_close_reclaims_the_other_requests_after_a_failed_reclaim( + scheduler: ApifyScheduler, + caplog: pytest.LogCaptureFixture, +) -> None: + """One failing reclaim does not stop the other in-flight requests from going back to the queue.""" + rq = cast('mock.MagicMock', scheduler._rq) + async_thread = cast('mock.MagicMock', scheduler._async_thread) + + apify_requests = [ + ApifyRequest( + url=f'https://example.com/{index}', + method='GET', + unique_key=f'https://example.com/{index}', + user_data={}, + ) + for index in range(2) + ] + + # The crawler stub keeps a reference to this set, so both requests stay busy as they are handed out. + busy: set[Request] = set() + scheduler._crawler = fake_crawler(downloader_busy=busy) + + async_thread.run_coro.side_effect = apify_requests + for _ in apify_requests: + busy.add(cast('Request', scheduler.next_request())) + + async_thread.run_coro.side_effect = [RuntimeError('boom'), None] + + with caplog.at_level(logging.ERROR, logger='apify.scrapy.scheduler'): + scheduler.close('shutdown') + + assert rq.reclaim_request.call_count == len(apify_requests) + errors = [record for record in caplog.records if record.levelno >= logging.ERROR] + assert len(errors) == 1 + + +def test_close_waits_for_the_non_blocking_updates(scheduler: ApifyScheduler) -> None: + """The event loop is not torn down before the updates fired off on the hot path have landed.""" + async_thread = cast('mock.MagicMock', scheduler._async_thread) + + scheduler.close('finished') + + calls = async_thread.mock_calls + assert calls.index(mock.call.wait_for_submitted()) < calls.index(mock.call.close()) + + +def test_a_failed_mark_keeps_the_request_tracked( + scheduler: ApifyScheduler, + caplog: pytest.LogCaptureFixture, +) -> None: + """A request whose mark-as-handled fails stays tracked, so the next resolution retries it.""" + async_thread = cast('mock.MagicMock', scheduler._async_thread) + + apify_request = ApifyRequest( + url='https://example.com', + method='GET', + unique_key='https://example.com', + user_data={}, + ) + async_thread.run_coro.return_value = apify_request + scheduler.next_request() + + scheduler._crawler = fake_crawler() + # The mark fails, then the queue reports itself unfinished because the request is still in progress. + async_thread.run_coro.side_effect = [RuntimeError('boom'), False] + + with caplog.at_level(logging.ERROR, logger='apify.scrapy.scheduler'): + assert scheduler.has_pending_requests() is True + + assert scheduler._requests_in_flight + errors = [record for record in caplog.records if record.levelno >= logging.ERROR] + assert len(errors) == 1 + + 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.""" monkeypatch.setattr('apify.scrapy.scheduler.is_asyncio_reactor_installed', lambda: True) From bc88a719cf4bf73dd4ee1ee2b318bf8a790c3f0c Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 21 Aug 2026 18:50:12 +0200 Subject: [PATCH 3/5] test(scrapy): cover the unique-key stamp round-trip --- .../unit/scrapy/requests/test_to_apify_request.py | 10 ++++++++++ .../scrapy/requests/test_to_scrapy_request.py | 15 +++++++++++++++ 2 files changed, 25 insertions(+) diff --git a/tests/unit/scrapy/requests/test_to_apify_request.py b/tests/unit/scrapy/requests/test_to_apify_request.py index 1ce15fe9..754e0290 100644 --- a/tests/unit/scrapy/requests/test_to_apify_request.py +++ b/tests/unit/scrapy/requests/test_to_apify_request.py @@ -190,6 +190,16 @@ def test_apify_request_id_in_meta_is_ignored(spider: Spider) -> None: assert apify_request.unique_key == 'https://example.com' +def test_unchanged_request_keeps_the_unique_key_it_was_stamped_with(spider: Spider) -> None: + """A request handed to Scrapy and enqueued again unchanged reuses the unique key it was minted for.""" + scrapy_request = to_scrapy_request(ApifyRequest.from_url('https://example.com'), spider) + + apify_request = to_apify_request(scrapy_request, spider) + + assert apify_request is not None + assert apify_request.unique_key == scrapy_request.meta['apify_request_unique_key'] + + def test_redirected_request_does_not_inherit_the_parents_unique_key(spider: Spider) -> None: """A redirect derived from a fetched request gets its own unique key instead of the parent's stamp.""" parent = to_scrapy_request(ApifyRequest.from_url('https://example.com/redirect'), spider) diff --git a/tests/unit/scrapy/requests/test_to_scrapy_request.py b/tests/unit/scrapy/requests/test_to_scrapy_request.py index 898312f2..c3803d7b 100644 --- a/tests/unit/scrapy/requests/test_to_scrapy_request.py +++ b/tests/unit/scrapy/requests/test_to_scrapy_request.py @@ -68,6 +68,21 @@ def test_without_reconstruction(spider: Spider) -> None: assert apify_request.unique_key == scrapy_request.meta.get('apify_request_unique_key') +def test_unique_key_is_stamped_together_with_its_url(spider: Spider) -> None: + """The queue's unique key is stamped alongside the URL it belongs to, so derived requests can be told apart.""" + apify_request = ApifyRequest( + url='https://example.com', + method='GET', + unique_key='https://example.com', + user_data={}, + ) + + scrapy_request = to_scrapy_request(apify_request, spider) + + assert scrapy_request.meta['apify_request_unique_key'] == apify_request.unique_key + assert scrapy_request.meta['apify_request_url'] == scrapy_request.url + + def test_without_reconstruction_with_optional_fields(spider: Spider) -> None: """The without-reconstruction path also carries optional headers and user data to the Scrapy request.""" apify_request = ApifyRequest( From 4ee5615380dd5d77e199df3ba23178e7c49474ef Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 21 Aug 2026 18:50:20 +0200 Subject: [PATCH 4/5] docs: explain what happens to Scrapy requests after a migration --- 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 49755de6..53d909e1 100644 --- a/docs/03_guides/06_scrapy.mdx +++ b/docs/03_guides/06_scrapy.mdx @@ -104,7 +104,7 @@ The following example shows a Scrapy Actor that scrapes page titles and enqueues ## Dealing with imminent migration to another host -Under some circumstances, the platform may decide to [migrate your Actor](https://docs.apify.com/academy/expert-scraping-with-apify/migrations-maintaining-state) from one piece of infrastructure to another while it's in progress. While [Crawlee](https://crawlee.dev/python)-based projects can pause and resume their work after a restart, achieving the same with a Scrapy-based project can be challenging. +Under some circumstances, the platform may decide to [migrate your Actor](https://docs.apify.com/academy/expert-scraping-with-apify/migrations-maintaining-state) from one piece of infrastructure to another while it's in progress. Requests that Scrapy hasn't finished when the run stops stay unhandled in the request queue, so the next run picks them up and downloads them from scratch. A Scrapy-based project doesn't resume where it left off the way a [Crawlee](https://crawlee.dev/python)-based one does, so items their callbacks already pushed can land in the dataset twice. As a workaround for this issue (tracked as [apify/actor-templates#303](https://github.com/apify/actor-templates/issues/303)), turn on caching with `HTTPCACHE_ENABLED` and set `HTTPCACHE_EXPIRATION_SECS` to at least a few minutes—the exact value depends on your use case. If your Actor gets migrated and restarted, the subsequent run will hit the cache, making it fast and avoiding unnecessary resource consumption. From 31391ed197b14376ca77a4ea453f9488126869ed Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 21 Aug 2026 19:15:13 +0200 Subject: [PATCH 5/5] refactor(scrapy): tighten comments and rename SUBMITTED_PRUNE_THRESHOLD --- src/apify/scrapy/_async_thread.py | 16 +++++------ src/apify/scrapy/requests.py | 13 ++++----- src/apify/scrapy/scheduler.py | 38 +++++++++++--------------- tests/unit/scrapy/test_async_thread.py | 4 +-- 4 files changed, 31 insertions(+), 40 deletions(-) diff --git a/src/apify/scrapy/_async_thread.py b/src/apify/scrapy/_async_thread.py index 3e3d57ee..d35dec3a 100644 --- a/src/apify/scrapy/_async_thread.py +++ b/src/apify/scrapy/_async_thread.py @@ -12,7 +12,7 @@ logger = getLogger(__name__) -_SUBMITTED_PRUNE_THRESHOLD = 128 +SUBMITTED_PRUNE_THRESHOLD = 128 """How many `submit_coro` futures may pile up before the finished ones are dropped from the tracking list.""" @@ -84,9 +84,8 @@ def submit_coro(self, coro: Coroutine) -> None: """Schedule a coroutine on the event loop without waiting for its result. Use this for work whose result nothing depends on, so the calling thread is not blocked by the round - trip. Failures are logged, as there is no caller left to propagate them to, and a coroutine still - pending when `close` runs is cancelled along with the rest - call `wait_for_submitted` before anything - that must not see that happen. + trip. Failures are logged, as no caller is left to propagate them to, and `close` cancels whatever is + still pending - call `wait_for_submitted` first if that matters. Args: coro: The coroutine to run. @@ -97,9 +96,9 @@ def submit_coro(self, coro: Coroutine) -> None: if self._eventloop.is_closed(): raise RuntimeError(f'The coroutine {coro} cannot be executed because the event loop is closed.') - # Drop the futures that already finished. `wait_for_submitted` only runs once Scrapy goes idle, so - # without this the list would hold every coroutine the whole crawl ever submitted, with its result. - if len(self._submitted) >= _SUBMITTED_PRUNE_THRESHOLD: + # `wait_for_submitted` only runs once Scrapy goes idle, so without pruning here the list would hold + # every coroutine the whole crawl ever submitted, with its result. + if len(self._submitted) >= SUBMITTED_PRUNE_THRESHOLD: self._submitted = [submitted for submitted in self._submitted if not submitted.done()] future = asyncio.run_coroutine_threadsafe(coro, self._eventloop) @@ -121,8 +120,7 @@ def wait_for_submitted(self, timeout: timedelta | None = None) -> None: self._submitted = list(futures.wait(self._submitted, timeout=timeout.total_seconds()).not_done) - # Returning with coroutines still pending breaks the guarantee the callers rely on, so say so rather - # than letting them act on effects that have not landed. + # Callers rely on the effects having landed, so a timeout has to be visible. if self._submitted: logger.warning(f'{len(self._submitted)} submitted coroutines did not finish within the timeout.') diff --git a/src/apify/scrapy/requests.py b/src/apify/scrapy/requests.py index e8333886..caa62413 100644 --- a/src/apify/scrapy/requests.py +++ b/src/apify/scrapy/requests.py @@ -76,11 +76,10 @@ def to_apify_request(scrapy_request: ScrapyRequest, spider: Spider) -> ApifyRequ try: if scrapy_request.dont_filter: request_kwargs['always_enqueue'] = True - # Reuse the queue's own unique key only while this is still the very request it was minted for. - # Scrapy derives new requests from a fetched one with `Request.replace()` (redirects) and spiders - # often forward `meta` verbatim to another URL; both inherit the stamp, and reusing it there would - # deduplicate the derived request against its parent and silently drop it. A stamp without a URL - # beside it was set by hand rather than by `to_scrapy_request`, so it is taken at face value. + # Reuse the queue's unique key only while this is still the request it was minted for. Redirects + # (`Request.replace()`) and spiders forwarding `meta` to another URL both inherit the stamp, and + # reusing it there deduplicates the derived request against its parent. A stamp without a URL beside + # it was set by hand, so it is taken at face value. elif (unique_key := scrapy_request.meta.get('apify_request_unique_key')) and ( scrapy_request.meta.get('apify_request_url', scrapy_request.url) == scrapy_request.url ): @@ -198,8 +197,8 @@ def to_scrapy_request(apify_request: ApifyRequest, spider: Spider) -> ScrapyRequ else: scrapy_request = ScrapyRequest(url=apify_request.url, method=apify_request.method) - # Stamp the queue's unique key together with the URL it belongs to, so that `to_apify_request` can tell - # this request apart from the ones Scrapy derives from it. + # Stamp the unique key together with the URL it belongs to, so `to_apify_request` can tell this request + # apart from the ones Scrapy derives from it. scrapy_request.meta['apify_request_unique_key'] = apify_request.unique_key scrapy_request.meta['apify_request_url'] = scrapy_request.url diff --git a/src/apify/scrapy/scheduler.py b/src/apify/scrapy/scheduler.py index 94096176..72f5656c 100644 --- a/src/apify/scrapy/scheduler.py +++ b/src/apify/scrapy/scheduler.py @@ -110,9 +110,8 @@ def close(self, reason: str) -> None: except Exception: logger.exception('Failed to resolve the requests still in flight in the request queue.') - # Whatever Scrapy did not finish - an interrupted run, an Actor migration - goes back to the queue - # while the event loop is still around, so the next run gets it as pending. Each request is - # reclaimed on its own, so one failure does not strand the rest. + # Whatever Scrapy did not finish goes back to the queue, so the next run gets it as pending. + # One failed reclaim must not strand the rest. for apify_request, _ in self._requests_in_flight: try: self._async_thread.run_coro(rq.reclaim_request(apify_request)) @@ -121,8 +120,8 @@ def close(self, reason: str) -> None: self._requests_in_flight.clear() - # Let the updates fired off on the hot path finish: closing the event loop cancels them silently, which - # would leave those requests unhandled in the queue. + # Closing the event loop cancels the updates fired off on the hot path silently, leaving those + # requests unhandled. self._async_thread.wait_for_submitted() try: @@ -152,8 +151,7 @@ def has_pending_requests(self) -> bool: # as in flight is provably finished. self._resolve_finished_requests(wait=True) - # The queue answers from its own bookkeeping, so an update still in flight would let it report itself - # finished while a request is unhandled - and closing the crawl would then cancel that update. + # The queue answers from its own bookkeeping, which a pending update has not reached yet. self._async_thread.wait_for_submitted() # Log here before re-raising: this coroutine ran on a separate event-loop thread, and the failure is @@ -207,8 +205,8 @@ def next_request(self) -> Request | None: if not isinstance(self._rq, RequestQueue): raise TypeError('self._rq must be an instance of the RequestQueue class') - # Resolve whatever Scrapy has finished since the last call. The engine polls this method throughout the - # crawl, which keeps the queue's view of progress current without blocking on the round trips. + # The engine polls this method throughout the crawl, so resolving here keeps the queue current + # without blocking on the round trips. self._resolve_finished_requests(wait=False) # Log here before re-raising: this coroutine ran on a separate event-loop thread, and the failure is @@ -225,9 +223,8 @@ def next_request(self) -> Request | None: if not isinstance(self.spider, Spider): raise TypeError('self.spider must be an instance of the Spider class') - # A malformed entry must not crash the whole run, so on failure it is logged and skipped rather than - # propagating. Such an unrecoverable entry (a corrupt or legacy payload) is marked as handled right - # away, otherwise the queue would keep handing it back forever. + # A corrupt or legacy payload must not crash the run, and is marked as handled right away, otherwise + # the queue would keep handing it back forever. try: scrapy_request = to_scrapy_request(apify_request, spider=self.spider) except Exception as exc: @@ -239,8 +236,8 @@ def next_request(self) -> Request | None: raise return None - # The entry stays unresolved in the queue until Scrapy is done with the request, so a run interrupted - # mid-flight leaves it pending instead of silently handled. + # The entry stays unresolved until Scrapy is done with the request, so a run interrupted mid-flight + # leaves it pending instead of silently handled. self._requests_in_flight.append((apify_request, scrapy_request)) return scrapy_request @@ -248,10 +245,9 @@ def next_request(self) -> Request | None: def _requests_busy_in_scrapy(self) -> set[Request]: """Return the requests Scrapy is still working on. - A request handed out by `next_request` joins the downloader's active set before the middleware chain - runs, and leaves the scraper's only once the spider callback and the item pipeline have finished with - it. Absence from both therefore means Scrapy is done with the request, whether it was downloaded, - dropped by a middleware or errored out. + A request joins the downloader's active set before the middleware chain runs and leaves the scraper's + only once the callback and the item pipeline are done, so absence from both means Scrapy has finished + with it - downloaded, dropped by a middleware or errored out alike. """ engine = self._crawler.engine if self._crawler is not None else None if engine is None: @@ -263,10 +259,8 @@ def _requests_busy_in_scrapy(self) -> set[Request]: def _resolve_finished_requests(self, *, wait: bool) -> None: """Mark every request Scrapy has finished processing as handled in the request queue. - Only a failure to dispatch the update - a timed-out or closed event loop - keeps a request tracked for - the next call to retry, without stopping the rest of the list from being resolved. The queue reports - the update's own failures by returning `None`, which is indistinguishable from success here, so such a - request is left unhandled for the next run to pick up. + A request whose update cannot be dispatched stays tracked for the next call to retry, without holding + up the rest of the list. Args: wait: Whether to block until the queue has been updated. Pass False on the crawl's hot path, where diff --git a/tests/unit/scrapy/test_async_thread.py b/tests/unit/scrapy/test_async_thread.py index 994db6e8..7030d51f 100644 --- a/tests/unit/scrapy/test_async_thread.py +++ b/tests/unit/scrapy/test_async_thread.py @@ -11,7 +11,7 @@ import pytest from ..._utils import poll_until_condition -from apify.scrapy._async_thread import _SUBMITTED_PRUNE_THRESHOLD, AsyncThread +from apify.scrapy._async_thread import SUBMITTED_PRUNE_THRESHOLD, AsyncThread async def _return(value: int) -> int: @@ -265,7 +265,7 @@ def test_submit_coro_drops_the_finished_futures() -> None: thread = AsyncThread() _wait_until_running(thread) - for _ in range(_SUBMITTED_PRUNE_THRESHOLD): + for _ in range(SUBMITTED_PRUNE_THRESHOLD): thread.submit_coro(_return(1)) assert futures.wait(list(thread._submitted), timeout=2).not_done == set()