fix(scrapy): keep traceback.print_exc() on background-loop coroutine errors

Author: vdusek

src/apify/scrapy/extensions/_httpcache.py

@@ -4,6 +4,7 @@
 import io
 import re
 import struct
+import traceback
 from datetime import timedelta
 from logging import getLogger
 from time import time
@@ -39,8 +40,8 @@ def __init__(self, settings: BaseSettings) -> None:
         # Upper bound on how many keys the per-spider-close cleanup sweeps (best-effort; `close_spider`).
         self._expiration_max_items: int = settings.getint('APIFY_HTTPCACHE_EXPIRATION_MAX_ITEMS', 100)
         self._expiration_secs: int = settings.getint('HTTPCACHE_EXPIRATION_SECS')
-        # Caps how long each coroutine run on the background event loop may take; defaults to 60 seconds.
         self._async_thread_timeout = timedelta(seconds=settings.getint('APIFY_ASYNC_THREAD_TIMEOUT_SECS', 60))
+        """Caps how long each coroutine run on the background event loop may take; defaults to 60 seconds."""
         self._spider: Spider | None = None
         self._kvs: KeyValueStore | None = None
         self._fingerprinter: RequestFingerprinterProtocol | None = None
@@ -67,7 +68,11 @@ async def open_kvs() -> KeyValueStore:
         logger.debug("Starting background thread for cache storage's event loop")
         self._async_thread = AsyncThread(default_timeout=self._async_thread_timeout)
         logger.debug(f"Opening cache storage's {kvs_name!r} key value store")
-        self._kvs = self._async_thread.run_coro(open_kvs())
+        try:
+            self._kvs = self._async_thread.run_coro(open_kvs())
+        except Exception:
+            traceback.print_exc()
+            raise
 
     def close_spider(self, _: Spider, current_time: int | None = None) -> None:
         """Close the cache storage for a spider."""
@@ -106,7 +111,11 @@ async def expire_kvs() -> None:
                             else:
                                 logger.debug(f'Valid cache item {item.key}')
 
-                self._async_thread.run_coro(expire_kvs())
+                try:
+                    self._async_thread.run_coro(expire_kvs())
+                except Exception:
+                    traceback.print_exc()
+                    raise
         finally:
             logger.debug('Closing cache storage')
             try:
@@ -128,7 +137,11 @@ def retrieve_response(self, _: Spider, request: Request, current_time: int | Non
             raise ValueError('Request fingerprinter not initialized')
 
         key = self._fingerprinter.fingerprint(request).hex()
-        value = self._async_thread.run_coro(self._kvs.get_value(key))
+        try:
+            value = self._async_thread.run_coro(self._kvs.get_value(key))
+        except Exception:
+            traceback.print_exc()
+            raise
 
         if value is None:
             logger.debug('Cache miss', extra={'request': request})
@@ -175,7 +188,11 @@ def store_response(self, _: Spider, request: Request, response: Response) -> Non
             'body': response.body,
         }
         value = to_gzip(data)
-        self._async_thread.run_coro(self._kvs.set_value(key, value))
+        try:
+            self._async_thread.run_coro(self._kvs.set_value(key, value))
+        except Exception:
+            traceback.print_exc()
+            raise
 
 
 def to_gzip(data: dict, mtime: int | None = None) -> bytes:

src/apify/scrapy/scheduler.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import traceback
 from datetime import timedelta
 from logging import getLogger
 from typing import TYPE_CHECKING
@@ -73,6 +74,7 @@ async def open_rq() -> RequestQueue:
             self._rq = self._async_thread.run_coro(open_rq())
         except Exception:
             self._async_thread.close()
+            traceback.print_exc()
             raise
 
         return None
@@ -107,7 +109,12 @@ def has_pending_requests(self) -> bool:
         if not isinstance(self._rq, RequestQueue):
             raise TypeError('self._rq must be an instance of the RequestQueue class')
 
-        is_finished = self._async_thread.run_coro(self._rq.is_finished())
+        try:
+            is_finished = self._async_thread.run_coro(self._rq.is_finished())
+        except Exception:
+            traceback.print_exc()
+            raise
+
         return not is_finished
 
     def enqueue_request(self, request: Request) -> bool:
@@ -135,7 +142,12 @@ def enqueue_request(self, request: Request) -> bool:
         if not isinstance(self._rq, RequestQueue):
             raise TypeError('self._rq must be an instance of the RequestQueue class')
 
-        result = self._async_thread.run_coro(self._rq.add_request(apify_request))
+        try:
+            result = self._async_thread.run_coro(self._rq.add_request(apify_request))
+        except Exception:
+            traceback.print_exc()
+            raise
+
         logger.debug(f'rq.add_request result: {result}')
         return not bool(result.was_already_present)
 
@@ -149,7 +161,12 @@ def next_request(self) -> Request | None:
         if not isinstance(self._rq, RequestQueue):
             raise TypeError('self._rq must be an instance of the RequestQueue class')
 
-        apify_request = self._async_thread.run_coro(self._rq.fetch_next_request())
+        try:
+            apify_request = self._async_thread.run_coro(self._rq.fetch_next_request())
+        except Exception:
+            traceback.print_exc()
+            raise
+
         logger.debug(f'Fetched apify_request: {apify_request}')
         if apify_request is None:
             return None
@@ -168,7 +185,11 @@ def next_request(self) -> 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.
-        self._async_thread.run_coro(self._rq.mark_request_as_handled(apify_request))
+        try:
+            self._async_thread.run_coro(self._rq.mark_request_as_handled(apify_request))
+        except Exception:
+            traceback.print_exc()
+            raise
 
         if scrapy_request is None:
             return None

tests/unit/scrapy/test_async_thread.py

@@ -3,26 +3,20 @@
 import asyncio
 import logging
 import threading
-import time
 from concurrent import futures
 from datetime import timedelta
 from typing import Any, Literal
 
 import pytest
 
+from ..._utils import poll_until_condition
 from apify.scrapy._async_thread import AsyncThread
 
 
 def _wait_until_running(thread: AsyncThread, timeout: float = 2.0) -> None:
     """Block until the background event loop is running, so `run_coro` does not race the thread startup."""
-    deadline = time.monotonic() + timeout
-    while not thread._eventloop.is_running():
-        if time.monotonic() > deadline:
-            raise AssertionError('The event loop did not start in time.')
-        time.sleep(0.01)
-
-
-# Coroutine execution
+    if not asyncio.run(poll_until_condition(thread._eventloop.is_running, timeout=timeout, poll_interval=0.01)):
+        raise AssertionError('The event loop did not start in time.')
 
 
 def test_run_coro_cancels_the_coroutine_on_timeout() -> None:
@@ -66,9 +60,6 @@ async def boom() -> None:
     assert [record for record in caplog.records if record.levelno >= logging.ERROR] == []
 
 
-# Shutdown
-
-
 def test_close_is_idempotent() -> None:
     """Calling `close` twice is a no-op the second time, not a `RuntimeError` on the closed loop."""
     thread = AsyncThread()

tests/unit/scrapy/test_scheduler.py

@@ -155,18 +155,20 @@ def test_next_request_returns_none_when_queue_empty(scheduler: ApifyScheduler) -
     rq.mark_request_as_handled.assert_not_called()
 
 
-def test_next_request_does_not_print_traceback_to_stderr(
+def test_next_request_prints_traceback_to_stderr(
     scheduler: ApifyScheduler,
     capsys: pytest.CaptureFixture[str],
 ) -> None:
-    """A failure propagates as-is, without `traceback.print_exc()` printing a second copy past the log formatter."""
+    """A failure in the coroutine run prints a traceback to stderr via `traceback.print_exc()` before propagating."""
     async_thread = cast('mock.MagicMock', scheduler._async_thread)
     async_thread.run_coro.side_effect = RuntimeError('boom')
 
     with pytest.raises(RuntimeError, match='boom'):
         scheduler.next_request()
 
-    assert capsys.readouterr().err == ''
+    captured = capsys.readouterr()
+    assert 'Traceback (most recent call last)' in captured.err
+    assert 'RuntimeError: boom' in captured.err
 
 
 def test_from_crawler_reads_async_thread_timeout_setting(monkeypatch: pytest.MonkeyPatch) -> None: