diff --git a/src/crawlee/_autoscaling/autoscaled_pool.py b/src/crawlee/_autoscaling/autoscaled_pool.py index 89f7c89312..fbbef06ddb 100644 --- a/src/crawlee/_autoscaling/autoscaled_pool.py +++ b/src/crawlee/_autoscaling/autoscaled_pool.py @@ -279,7 +279,9 @@ async def _worker_task(self) -> None: timeout=self._TASK_TIMEOUT.total_seconds() if self._TASK_TIMEOUT is not None else None, ) except asyncio.TimeoutError: - timeout_str = self._TASK_TIMEOUT.total_seconds() if self._TASK_TIMEOUT is not None else '*not set*' - logger.warning(f'Task timed out after {timeout_str} seconds') + if self._TASK_TIMEOUT is None: + # Without a timeout, `wait_for` cannot time out - the error comes from the task function itself. + raise + logger.warning(f'Task timed out after {self._TASK_TIMEOUT.total_seconds()} seconds') finally: logger.debug('Worker task finished') diff --git a/src/crawlee/_utils/recurring_task.py b/src/crawlee/_utils/recurring_task.py index 99f21499cb..99d7048a2c 100644 --- a/src/crawlee/_utils/recurring_task.py +++ b/src/crawlee/_utils/recurring_task.py @@ -25,11 +25,8 @@ class RecurringTask: """ def __init__(self, func: Callable, delay: timedelta) -> None: - logger.debug( - 'Calling RecurringTask.__init__(func={%s}, delay={%s})...', - func.__name__ if hasattr(func, '__name__') else func.__class__.__name__, - delay, - ) + self._func_name = func.__name__ if hasattr(func, '__name__') else func.__class__.__name__ + logger.debug('Calling RecurringTask.__init__(func={%s}, delay={%s})...', self._func_name, delay) self.func = func self.delay = delay self.task: asyncio.Task | None = None @@ -49,20 +46,22 @@ async def __aexit__( async def _wrapper(self) -> None: """Continuously execute the provided function with the specified delay. - Run the function in a loop, waiting for the configured delay between executions. - Supports both synchronous and asynchronous functions. + Run the function in a loop, waiting for the configured delay between executions. Supports both synchronous and + asynchronous functions. An exception raised by the function is logged and does not stop the recurrence. """ sleep_time_secs = self.delay.total_seconds() while True: - await self.func() if inspect.iscoroutinefunction(self.func) else self.func() + try: + await self.func() if inspect.iscoroutinefunction(self.func) else self.func() + except Exception: + logger.exception('Recurring task `%s` failed; it will run again after the delay.', self._func_name) await asyncio.sleep(sleep_time_secs) def start(self) -> None: """Start the recurring task execution.""" - name = self.func.__name__ if hasattr(self.func, '__name__') else self.func.__class__.__name__ self.task = asyncio.create_task( self._wrapper(), - name=f'Task-recurring-{name}', + name=f'Task-recurring-{self._func_name}', ) async def stop(self) -> None: diff --git a/tests/unit/_autoscaling/test_autoscaled_pool.py b/tests/unit/_autoscaling/test_autoscaled_pool.py index c77a1d8926..5af4e4b8e7 100644 --- a/tests/unit/_autoscaling/test_autoscaled_pool.py +++ b/tests/unit/_autoscaling/test_autoscaled_pool.py @@ -145,6 +145,31 @@ async def run() -> None: await pool.run() +async def test_propagates_timeout_error_from_task_function(system_status: SystemStatus | Mock) -> None: + """Test that a `TimeoutError` raised by the task function propagates instead of being swallowed by the pool.""" + started_count = 0 + + async def run() -> None: + nonlocal started_count + started_count += 1 + raise asyncio.TimeoutError('Fetching a session from the pool timed out after 10 seconds') + + pool = AutoscaledPool( + system_status=system_status, + run_task_function=run, + is_task_ready_function=lambda: future(True), + is_finished_function=lambda: future(started_count > 0), + concurrency_settings=ConcurrencySettings( + min_concurrency=1, + desired_concurrency=1, + max_concurrency=1, + ), + ) + + with pytest.raises(asyncio.TimeoutError, match=r'Fetching a session from the pool timed out'): + await pool.run() + + async def test_autoscales( monkeypatch: pytest.MonkeyPatch, system_status: SystemStatus | Mock, diff --git a/tests/unit/_utils/test_recurring_task.py b/tests/unit/_utils/test_recurring_task.py index d2ddd2d861..0e347b0b28 100644 --- a/tests/unit/_utils/test_recurring_task.py +++ b/tests/unit/_utils/test_recurring_task.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import logging from datetime import timedelta from unittest.mock import AsyncMock @@ -53,3 +54,29 @@ async def test_execution(function: AsyncMock, delay: timedelta) -> None: assert task.func.call_count >= 3 await task.stop() + + +async def test_execution_continues_after_exception(delay: timedelta, caplog: pytest.LogCaptureFixture) -> None: + """Test that the recurring task logs an exception raised by its function and keeps executing.""" + caplog.set_level(logging.ERROR, logger='crawlee._utils.recurring_task') + call_count = 0 + second_call_done = asyncio.Event() + + async def func() -> None: + nonlocal call_count + call_count += 1 + if call_count == 1: + raise ValueError('Scheduled crash') + second_call_done.set() + + task = RecurringTask(func, delay) + task.start() + + await asyncio.wait_for(second_call_done.wait(), timeout=5) + await task.stop() + + assert call_count >= 2 + assert any( + record.name == 'crawlee._utils.recurring_task' and record.levelno == logging.ERROR and record.exc_info + for record in caplog.records + )