fix(throttling-manager): Stop holding a concurrency slot while a domain is throttled

docs/guides/request_throttling.mdx

@@ -18,7 +18,7 @@ The <ApiLink to="class/ThrottlingRequestManager">`ThrottlingRequestManager`</Api
 - **Routes requests** for listed domains into dedicated sub-managers at insertion time.
 - **Enforces delays** from HTTP 429 responses (exponential backoff) and `robots.txt` crawl-delay directives.
 - **Schedules fairly** by fetching from the domain that has been waiting the longest.
-- **Sleeps intelligently** when all configured domains are throttled, instead of busy-waiting.
+- **Releases the concurrency slot** when all configured domains are throttled, instead of holding it for the whole cooldown.
 
 Requests for domains **not** in the configured list pass through to the main queue without any throttling.
 
@@ -40,6 +40,8 @@ To use request throttling, create a <ApiLink to="class/ThrottlingRequestManager"
 
 4. **Fair scheduling**: `fetch_next_request` sorts available sub-managers by how long each domain has been waiting, ensuring no domain is starved.
 
+5. **Cooldown handling**: While a domain is in a cooldown, its queued requests don't count as dispatchable, so the crawler's autoscaled pool idles instead of keeping a worker slot blocked. The requests still count towards completion, so the crawl waits for them and finishes only once every one has been handled.
+
 :::tip
 
 The `ThrottlingRequestManager` is an opt-in feature. If you don't pass it to your crawler, requests are processed normally without any per-domain throttling.

src/crawlee/request_loaders/_throttling_request_manager.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import asyncio
-import contextlib
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from logging import getLogger
@@ -38,9 +37,11 @@ class ThrottlingRequestManager(RequestManager, Generic[TRequestManager]):
     Requests for explicitly configured domains are routed into dedicated sub-managers at insertion time — each request
     lives in exactly one manager, eliminating duplication and simplifying deduplication.
 
-    When `fetch_next_request()` is called, it returns requests from the sub-manager whose domain has been waiting the
-    longest. If all configured domains are throttled, it falls back to the inner manager for non-throttled domains. If
-    the inner manager is also empty and all sub-managers are throttled, it sleeps until the earliest cooldown expires.
+    `fetch_next_request()` takes from the sub-manager whose domain has been waiting the longest, skipping domains in a
+    cooldown, and falls back to the inner manager when no sub-manager yields a request. If nothing can be dispatched
+    right now, it returns `None` rather than waiting, so the caller's task slot is released. `is_empty()` reports the
+    same view and reads as empty while every remaining request sits in a cooldown, whereas `is_finished()` counts those
+    requests, so the crawl idles until they are dispatchable instead of ending early.
 
     Delay sources:
     - HTTP 429 responses (via `record_domain_delay`)
@@ -101,9 +102,6 @@ def __init__(
         self._request_manager_opener = request_manager_opener
         self._domain_states: dict[str, _DomainState] = {d.lower(): _DomainState(domain=d.lower()) for d in domains if d}
         self._sub_managers: dict[str, TRequestManager] = {}
-        self._new_work_event = asyncio.Event()
-        """Set whenever a request is added or reclaimed. Lets `fetch_next_request` wake from a throttle
-        wait early when fresh work appears, instead of sleeping for the full computed cooldown."""
 
     @property
     def inner(self) -> TRequestManager:
@@ -140,12 +138,9 @@ async def add_request(self, request: str | Request, *, forefront: bool = False)
 
         if domain in self._domain_states:
             sm = await self._get_or_create_sub_manager(domain)
-            result = await sm.add_request(request, forefront=forefront)
-        else:
-            result = await self._inner.add_request(request, forefront=forefront)
+            return await sm.add_request(request, forefront=forefront)
 
-        self._signal_new_work()
-        return result
+        return await self._inner.add_request(request, forefront=forefront)
 
     @override
     async def add_requests(
@@ -192,68 +187,31 @@ async def add_requests(
                 wait_for_all_requests_to_be_added_timeout=wait_for_all_requests_to_be_added_timeout,
             )
 
-        if inner_requests or domain_requests:
-            self._signal_new_work()
-
     @override
     async def fetch_next_request(self) -> Request | None:
         """Fetch the next request, respecting per-domain delays.
 
-        Sub-managers are checked in order of longest-overdue domain first (sorted by `throttled_until` ascending). If
-        all configured domains are throttled, falls back to the inner manager for non-throttled domains. If the inner
-        manager is also empty and all sub-managers are throttled, waits until either the earliest domain becomes
-        available or new work is added (whichever comes first).
-        """
-        while True:
-            # Clear the event before checking the queues. Any add/reclaim that races with this iteration will set the
-            # event again, so the wait at the end of the loop returns immediately rather than blocking until the
-            # throttle expires.
-            self._new_work_event.clear()
-
-            now = datetime.now(timezone.utc)
-            available_domains = sorted(
-                (
-                    domain
-                    for domain, state in self._domain_states.items()
-                    if domain in self._sub_managers and now >= state.throttled_until
-                ),
-                key=lambda d: self._domain_states[d].throttled_until,
-            )
-
-            for domain in available_domains:
-                req = await self._sub_managers[domain].fetch_next_request()
-                if req:
-                    self._mark_domain_dispatched(domain)
-                    return req
+        Sub-managers are checked in order of longest-overdue domain first, then the inner manager. Domains in a
+        cooldown are skipped, so the call returns `None` when nothing is dispatchable right now.
 
-            request = await self._inner.fetch_next_request()
+        Note:
+            Unlike the `RequestLoader.fetch_next_request` contract, a `None` result does not imply that `is_finished()`
+            is `True` - it only means nothing is dispatchable right now. Since the manager never waits out a cooldown
+            itself, the dispatch cadence is only as precise as the caller's polling interval: a cooldown expiring
+            between two polls is picked up on the next one.
+        """
+        for domain in self._fetchable_domains():
+            request = await self._sub_managers[domain].fetch_next_request()
             if request is not None:
+                self._mark_domain_dispatched(domain)
                 return request
 
-            if not self._sub_managers:
-                return None
-
-            sub_managers_empty = await asyncio.gather(*(sm.is_empty() for sm in self._sub_managers.values()))
-            if all(sub_managers_empty):
-                return None
-
-            earliest = self._get_earliest_available_time(now)
-            sleep_duration = max(
-                (earliest - now).total_seconds(),
-                0.1,  # Avoid tight loops if a throttle expired during the previous iteration.
-            )
-            logger.debug(
-                f'All configured domains are throttled and inner manager is empty. '
-                f'Waiting up to {sleep_duration:.1f}s for earliest domain to become available or new work.'
-            )
-            await self._wait_for_new_work_or_timeout(sleep_duration)
+        return await self._inner.fetch_next_request()
 
     @override
     async def reclaim_request(self, request: Request, *, forefront: bool = False) -> ProcessedRequest | None:
         manager = self._select_manager(request.url)
-        result = await manager.reclaim_request(request, forefront=forefront)
-        self._signal_new_work()
-        return result
+        return await manager.reclaim_request(request, forefront=forefront)
 
     @override
     async def mark_request_as_handled(self, request: Request) -> ProcessedRequest | None:
@@ -278,7 +236,14 @@ async def get_total_count(self) -> int:
 
     @override
     async def is_empty(self) -> bool:
-        results = await asyncio.gather(self._inner.is_empty(), *(sm.is_empty() for sm in self._sub_managers.values()))
+        """Report whether anything can be dispatched right now.
+
+        Requests queued for a domain in a cooldown do not count. They still count towards `is_finished`, so the crawl
+        waits for them.
+        """
+        results = await asyncio.gather(
+            self._inner.is_empty(), *(self._sub_managers[d].is_empty() for d in self._fetchable_domains())
+        )
         return all(results)
 
     @override
@@ -385,15 +350,18 @@ def _is_domain_throttled(self, domain: str) -> bool:
             return False
         return datetime.now(timezone.utc) < state.throttled_until
 
-    def _get_earliest_available_time(self, now: datetime) -> datetime:
-        """Get the earliest time any throttled domain becomes available."""
-        earliest = now + self._max_delay
-
-        for state in self._domain_states.values():
-            if now < state.throttled_until < earliest:
-                earliest = state.throttled_until
-
-        return earliest
+    def _fetchable_domains(self) -> list[str]:
+        """Return the configured domains that are not in a cooldown right now, longest-overdue first."""
+        now = datetime.now(timezone.utc)
+        available = [
+            domain
+            for domain, state in self._domain_states.items()
+            # Every configured domain has state from construction, but sub-managers are created lazily on first
+            # insertion, so this check keeps the `_sub_managers[domain]` lookups in the callers safe.
+            if domain in self._sub_managers and now >= state.throttled_until
+        ]
+        available.sort(key=lambda domain: self._domain_states[domain].throttled_until)
+        return available
 
     def _mark_domain_dispatched(self, domain: str) -> None:
         """Record that a request to this domain was just dispatched.
@@ -404,27 +372,13 @@ def _mark_domain_dispatched(self, domain: str) -> None:
         if state is not None and state.crawl_delay is not None:
             state.throttled_until = datetime.now(timezone.utc) + state.crawl_delay
 
-    def _signal_new_work(self) -> None:
-        """Wake `fetch_next_request` if it is sleeping inside a throttle wait."""
-        self._new_work_event.set()
-
     def _select_manager(self, url: str) -> TRequestManager:
         """Return the manager that owns the given URL — its sub-manager if one exists, otherwise the inner."""
         domain = self._extract_domain(url)
         if domain in self._sub_managers:
             return self._sub_managers[domain]
         return self._inner
 
-    async def _wait_for_new_work_or_timeout(self, timeout: float) -> None:
-        """Wait until new work is signaled or `timeout` seconds elapse, whichever comes first.
-
-        The signal is set by `add_request`, `add_requests`, and `reclaim_request`, allowing `fetch_next_request` to wake
-        up immediately when fresh work appears during a throttle wait instead of sleeping for the full computed cooldown
-        (up to `max_delay`).
-        """
-        with contextlib.suppress(asyncio.TimeoutError):
-            await asyncio.wait_for(self._new_work_event.wait(), timeout=timeout)
-
 
 class _RequestManagerOpener(Protocol[TRequestManager]):
     """Callable that opens a `RequestManager` instance.

tests/unit/crawlers/_basic/test_basic_crawler.py

@@ -2502,3 +2502,38 @@ async def test_warn_unconfigured_throttle_domain_once_per_domain(caplog: pytest.
     assert len(matching) == 2
     assert any('a.example.com' in r.getMessage() for r in matching)
     assert any('other.example.com' in r.getMessage() for r in matching)
+
+
+async def test_throttled_domain_waits_out_backoff_without_ending_the_crawl() -> None:
+    """A domain in a cooldown reads as empty to the autoscaled pool, yet its queued requests are still crawled."""
+    storage_client = MemoryStorageClient()
+    # The throttler opens its sub-managers through the global service locator, so point that at the same client.
+    service_locator.set_storage_client(storage_client)
+    inner = await RequestQueue.open(name='test-inner-backoff', storage_client=storage_client)
+    throttler = ThrottlingRequestManager(
+        inner,
+        domains=['throttled.placeholder.com'],
+        request_manager_opener=RequestQueue.open,
+    )
+    # A single worker slot makes the ordering deterministic: the second dispatch cannot start before the first
+    # handler has armed the backoff.
+    crawler = BasicCrawler(
+        request_manager=throttler,
+        configure_logging=False,
+        concurrency_settings=ConcurrencySettings(desired_concurrency=1, max_concurrency=1),
+    )
+    dispatched_at = list[float]()
+    empty_during_cooldown = list[bool]()
+
+    @crawler.router.default_handler
+    async def handler(context: BasicCrawlingContext) -> None:
+        dispatched_at.append(time.monotonic())
+        if len(dispatched_at) == 1:
+            throttler.record_domain_delay(context.request.url, retry_after=timedelta(milliseconds=500))
+            empty_during_cooldown.append(await throttler.is_empty())
+
+    await crawler.run(['https://throttled.placeholder.com/a', 'https://throttled.placeholder.com/b'])
+
+    assert empty_during_cooldown == [True]
+    assert len(dispatched_at) == 2
+    assert dispatched_at[1] - dispatched_at[0] >= 0.5