fix: prevent silent RecurringTask death and swallowed timeouts in AutoscaledPool

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')

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:

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,

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
+    )