fix: Fix incorrect use of desired concurrency ratio (apify#780)

### Description
Concurrency ratio was not correctly used in Python version of autoscaled
pool. Align it with the Javascript implementation of autoscaled pool.
Add test.
### Issues

- Closes: apify#759

---------

Co-authored-by: Vlada Dusek <[email protected]>

Author: 2 people

src/crawlee/_autoscaling/autoscaled_pool.py

@@ -195,7 +195,7 @@ def _autoscale(self) -> None:
         """Inspect system load status and adjust desired concurrency if necessary. Do not call directly."""
         status = self._system_status.get_historical_system_info()
 
-        min_current_concurrency = math.floor(self._desired_concurrency_ratio * self.current_concurrency)
+        min_current_concurrency = math.floor(self._desired_concurrency_ratio * self.desired_concurrency)
         should_scale_up = (
             status.is_system_idle
             and self._desired_concurrency < self._max_concurrency

tests/unit/_autoscaling/test_autoscaled_pool.py

@@ -5,6 +5,7 @@
 import asyncio
 from contextlib import suppress
 from datetime import datetime, timedelta, timezone
+from itertools import chain, repeat
 from typing import TYPE_CHECKING, TypeVar, cast
 from unittest.mock import Mock
 
@@ -208,6 +209,60 @@ def get_historical_system_info() -> SystemInfo:
             await pool_run_task
 
 
+async def test_autoscales_uses_desired_concurrency_ratio(system_status: SystemStatus | Mock) -> None:
+    """Test that desired concurrency ratio can limit desired concurrency.
+
+    This test creates situation where only one task is ready and then no other task is ever ready.
+    This creates situation where the system could scale up desired concurrency, but it will not do so because
+    desired_concurrency_ratio=1 means that first the system would have to increase current concurrency to same number as
+    desired concurrency and due to no other task ever being ready, it will never happen. Thus desired concurrency will
+    stay 2 as was the initial setup, even though other conditions would allow the increase. (max_concurrency=4,
+    system being idle)."""
+
+    async def run() -> None:
+        await asyncio.sleep(0.1)
+
+    is_task_ready_iterator = chain([future(True)], repeat(future(False)))
+
+    def is_task_ready_function() -> Awaitable[bool]:
+        return next(is_task_ready_iterator)
+
+    def get_historical_system_info() -> SystemInfo:
+        return SystemInfo(
+            cpu_info=LoadRatioInfo(limit_ratio=0.9, actual_ratio=0.3),
+            memory_info=LoadRatioInfo(limit_ratio=0.9, actual_ratio=0.3),
+            event_loop_info=LoadRatioInfo(limit_ratio=0.9, actual_ratio=0.3),
+            client_info=LoadRatioInfo(limit_ratio=0.9, actual_ratio=0.3),
+        )
+
+    cast(Mock, system_status.get_historical_system_info).side_effect = get_historical_system_info
+
+    pool = AutoscaledPool(
+        system_status=system_status,
+        run_task_function=run,
+        is_task_ready_function=is_task_ready_function,
+        is_finished_function=lambda: future(False),
+        concurrency_settings=ConcurrencySettings(
+            min_concurrency=2,
+            desired_concurrency=2,
+            max_concurrency=4,
+        ),
+        autoscale_interval=timedelta(seconds=0.1),
+        desired_concurrency_ratio=1,
+    )
+
+    pool_run_task = asyncio.create_task(pool.run(), name='pool run task')
+    try:
+        for _ in range(5):
+            assert pool.desired_concurrency == 2
+            await asyncio.sleep(0.1)
+
+    finally:
+        pool_run_task.cancel()
+        with suppress(asyncio.CancelledError):
+            await pool_run_task
+
+
 async def test_max_tasks_per_minute_works(system_status: SystemStatus | Mock) -> None:
     done_count = 0