fix(throttling-manager): Open the per-domain sub-queues on first use instead of on insert

docs/guides/request_throttling.mdx

@@ -40,6 +40,21 @@ 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.
 
+## Sub-manager storage
+
+Each configured domain gets its own sub-manager, opened through the `request_manager_opener` callback under the alias `throttled-<domain>`. All of them are opened the first time you use the manager, so a domain that never receives a request still gets an empty store.
+
+Opening the sub-managers up front also makes requests that a previous run left behind visible again. Whether they're resumed or discarded depends on <ApiLink to="class/Configuration#purge_on_start">`Configuration.purge_on_start`</ApiLink>:
+
+- With the default `purge_on_start=True`, the leftover requests are purged when the sub-manager opens, just like the requests in an unnamed inner queue.
+- With `purge_on_start=False`, the leftover requests are picked up and crawled.
+
+:::warning
+
+Named storages are exempt from `purge_on_start`, but aliased ones aren't. If you give the inner <ApiLink to="class/RequestQueue">`RequestQueue`</ApiLink> a `name` to make it persistent, the inner queue keeps its requests across a restart while the per-domain stores are still purged. To keep the requests in both, set `purge_on_start=False`.
+
+:::
+
 :::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

@@ -35,8 +35,10 @@
 class ThrottlingRequestManager(RequestManager, Generic[TRequestManager]):
     """A request manager that wraps another and enforces per-domain delays.
 
-    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.
+    Requests for explicitly configured domains are routed into dedicated sub-managers. A request added through this
+    manager lives in exactly one of them, which keeps deduplication within a single store. A request that reached
+    `inner` before its domain was configured stays there, and is fetched and completed against `inner` without the
+    domain's delay applied.
 
     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
@@ -46,10 +48,17 @@ class ThrottlingRequestManager(RequestManager, Generic[TRequestManager]):
     - HTTP 429 responses (via `record_domain_delay`)
     - robots.txt crawl-delay directives (via `set_crawl_delay`)
 
-    The class is generic over the wrapped manager type. The `request_manager_opener` callback is used to construct
-    per-domain sub-managers at insertion time, so every sub-manager shares the same `RequestManager` subclass and
-    backing store as `inner`. The opener must accept `alias`, `storage_client`, and `configuration` keyword arguments
-    (as `RequestQueue.open` does) and return the same concrete subclass as `inner`.
+    The class is generic over the wrapped manager type. The first asynchronous operation - adding, fetching,
+    completing, counting, purging, or dropping - makes the `request_manager_opener` callback open one sub-manager per
+    configured domain, so every sub-manager shares the same `RequestManager` subclass and backing store as `inner`. The
+    synchronous delay methods (`record_domain_delay`, `record_success`, `set_crawl_delay`) only touch in-memory state
+    and never open anything. The opener must accept `alias`, `storage_client`, and `configuration` keyword arguments (as
+    `RequestQueue.open` does) and return the same concrete subclass as `inner`.
+
+    Opening the sub-managers up front also makes requests left over in a persistent store by a previous run visible
+    again. With the default `purge_on_start=True` those leftovers are purged at open, so resuming them requires
+    `purge_on_start=False`. Aliased stores are not exempt from that purge but named ones are, so a named `inner` keeps
+    its requests across a restart while the per-domain stores are emptied.
 
     ### Usage
 
@@ -86,9 +95,9 @@ def __init__(
             domains: Explicit list of domain hostnames to throttle. Only requests matching these domains will be routed
                 to per-domain sub-managers. Matching is case-insensitive (hostnames are lowercased) and exact: subdomain
                 wildcards such as `*.example.com` are not supported — list each subdomain explicitly if needed.
-            request_manager_opener: Async callable used to create per-domain sub-managers at insertion time. Must
-                accept `alias`, `storage_client`, and `configuration` keyword arguments and return the same concrete
-                subclass as `inner` (e.g. `RequestQueue.open` when `inner` is a `RequestQueue`).
+            request_manager_opener: Async callable used to open one sub-manager per configured domain on first use.
+                Must accept `alias`, `storage_client`, and `configuration` keyword arguments and return the same
+                concrete subclass as `inner` (e.g. `RequestQueue.open` when `inner` is a `RequestQueue`).
             service_locator: Service locator for creating sub-managers. If not provided, defaults to the global service
                 locator, ensuring consistency with the crawler's storage backend.
             base_delay: Initial delay after the first 429 response from a domain.
@@ -101,6 +110,21 @@ 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._sub_managers_ready = False
+        self._sub_managers_lock = asyncio.Lock()
+        self._in_flight_from_inner: set[tuple[str, str]] = set()
+        """`(unique_key, url)` pairs of configured-domain requests handed out by `fetch_next_request` from the inner
+        manager. Such a request can live in `inner` if it was added before its domain was listed, and it must be given
+        back to the manager it came from. Requests for unconfigured domains need no record, as they route to `inner` by
+        default. The URL is part of the key because `unique_key` may be set explicitly and is only unique per store, so
+        a key alone could match a same-key request held by a sub-manager.
+
+        The pair identifies a request by value, not by object. One URL can be in flight from both `inner` and its
+        sub-manager at once - deduplication is per store, so both may hold it - and the two completions can then be
+        routed to each other's manager. Both stores hold the key, so each completion still lands: the cost is a
+        duplicate crawl of that URL and a retry that skips the domain's delay, not a stalled queue. Telling the two
+        copies apart would take per-request identity, which `Request` cannot offer as it is unhashable and compares
+        by value."""
         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."""
@@ -112,18 +136,22 @@ def inner(self) -> TRequestManager:
 
     @override
     async def drop(self) -> None:
+        await self._ensure_sub_managers()
         await asyncio.gather(self._inner.drop(), *(sm.drop() for sm in self._sub_managers.values()))
         self._sub_managers.clear()
+        self._sub_managers_ready = False
+        self._in_flight_from_inner.clear()
 
     @override
     async def purge(self) -> None:
         """Empty the inner manager and all sub-managers, and reset transient per-domain throttle state.
 
-        The configured domain list and any robots.txt-derived `crawl_delay` are preserved; only the dynamic backoff
-        state (consecutive 429 counter and `throttled_until`) is cleared. Sub-managers are kept around so they don't
-        need to be re-opened on the next request — they're just emptied.
+        The configured domain list and any robots.txt-derived `crawl_delay` are preserved. Only the dynamic backoff
+        state (consecutive 429 counter and `throttled_until`) is cleared. Sub-managers stay open; they're just emptied.
         """
+        await self._ensure_sub_managers()
         await asyncio.gather(self._inner.purge(), *(sm.purge() for sm in self._sub_managers.values()))
+        self._in_flight_from_inner.clear()
         for state in self._domain_states.values():
             state.consecutive_429_count = 0
             state.throttled_until = _NEVER_THROTTLED
@@ -135,12 +163,13 @@ async def add_request(self, request: str | Request, *, forefront: bool = False)
         Requests for explicitly configured domains are routed directly to their per-domain sub-manager. All other
         requests go to the inner manager.
         """
+        await self._ensure_sub_managers()
+
         url = self._get_url_from_request(request)
         domain = self._extract_domain(url)
 
         if domain in self._domain_states:
-            sm = await self._get_or_create_sub_manager(domain)
-            result = await sm.add_request(request, forefront=forefront)
+            result = await self._sub_managers[domain].add_request(request, forefront=forefront)
         else:
             result = await self._inner.add_request(request, forefront=forefront)
 
@@ -159,6 +188,8 @@ async def add_requests(
         wait_for_all_requests_to_be_added_timeout: timedelta | None = None,
     ) -> None:
         """Add multiple requests, routing each to the appropriate manager."""
+        await self._ensure_sub_managers()
+
         inner_requests: list[str | Request] = []
         domain_requests: dict[str, list[str | Request]] = {}
 
@@ -182,8 +213,7 @@ async def add_requests(
             )
 
         for domain, reqs in domain_requests.items():
-            sm = await self._get_or_create_sub_manager(domain)
-            await sm.add_requests(
+            await self._sub_managers[domain].add_requests(
                 reqs,
                 forefront=forefront,
                 batch_size=batch_size,
@@ -204,6 +234,8 @@ async def fetch_next_request(self) -> Request | None:
         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).
         """
+        await self._ensure_sub_managers()
+
         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
@@ -212,11 +244,7 @@ async def fetch_next_request(self) -> Request | None:
 
             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
-                ),
+                (domain for domain, state in self._domain_states.items() if now >= state.throttled_until),
                 key=lambda d: self._domain_states[d].throttled_until,
             )
 
@@ -228,11 +256,10 @@ async def fetch_next_request(self) -> Request | None:
 
             request = await self._inner.fetch_next_request()
             if request is not None:
+                if self._extract_domain(request.url) in self._domain_states:
+                    self._in_flight_from_inner.add((request.unique_key, request.url))
                 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
@@ -250,39 +277,47 @@ async def fetch_next_request(self) -> Request | None:
 
     @override
     async def reclaim_request(self, request: Request, *, forefront: bool = False) -> ProcessedRequest | None:
-        manager = self._select_manager(request.url)
+        await self._ensure_sub_managers()
+        manager = self._fetch_owner(request)
         result = await manager.reclaim_request(request, forefront=forefront)
+        self._clear_fetch_owner(request)
         self._signal_new_work()
         return result
 
     @override
     async def mark_request_as_handled(self, request: Request) -> ProcessedRequest | None:
-        manager = self._select_manager(request.url)
+        await self._ensure_sub_managers()
+        manager = self._fetch_owner(request)
         result = await manager.mark_request_as_handled(request)
+        self._clear_fetch_owner(request)
         self.record_success(request.url)
         return result
 
     @override
     async def get_handled_count(self) -> int:
+        await self._ensure_sub_managers()
         counts = await asyncio.gather(
             self._inner.get_handled_count(), *(sm.get_handled_count() for sm in self._sub_managers.values())
         )
         return sum(counts)
 
     @override
     async def get_total_count(self) -> int:
+        await self._ensure_sub_managers()
         counts = await asyncio.gather(
             self._inner.get_total_count(), *(sm.get_total_count() for sm in self._sub_managers.values())
         )
         return sum(counts)
 
     @override
     async def is_empty(self) -> bool:
+        await self._ensure_sub_managers()
         results = await asyncio.gather(self._inner.is_empty(), *(sm.is_empty() for sm in self._sub_managers.values()))
         return all(results)
 
     @override
     async def is_finished(self) -> bool:
+        await self._ensure_sub_managers()
         results = await asyncio.gather(
             self._inner.is_finished(), *(sm.is_finished() for sm in self._sub_managers.values())
         )
@@ -368,15 +403,39 @@ def _get_domain_state(self, url: str) -> _DomainState | None:
         domain = self._extract_domain(url)
         return self._domain_states.get(domain) if domain else None
 
-    async def _get_or_create_sub_manager(self, domain: str) -> TRequestManager:
-        """Get or create a per-domain sub-manager using the configured `request_manager_opener`."""
-        if domain not in self._sub_managers:
-            self._sub_managers[domain] = await self._request_manager_opener(
-                alias=f'throttled-{domain}',
-                storage_client=self._service_locator.get_storage_client(),
-                configuration=self._service_locator.get_configuration(),
+    async def _open_sub_manager(self, domain: str) -> None:
+        """Open the sub-manager for a single domain using the configured `request_manager_opener`."""
+        self._sub_managers[domain] = await self._request_manager_opener(
+            alias=f'throttled-{domain}',
+            storage_client=self._service_locator.get_storage_client(),
+            configuration=self._service_locator.get_configuration(),
+        )
+
+    async def _ensure_sub_managers(self) -> None:
+        """Open a sub-manager for every configured domain, once.
+
+        Sub-managers that opened before a sibling failed are kept, so a retry after a failure opens only what is
+        still missing.
+        """
+        if self._sub_managers_ready:
+            return
+
+        async with self._sub_managers_lock:
+            if self._sub_managers_ready:
+                return
+
+            # Every attempt has to settle before the lock is released. A propagating error would leave the remaining
+            # openers running unawaited, free to write into `_sub_managers` after a retry has already replaced the
+            # manager for that domain - stranding whatever the loser of that race holds.
+            missing = [domain for domain in self._domain_states if domain not in self._sub_managers]
+            results = await asyncio.gather(
+                *(self._open_sub_manager(domain) for domain in missing), return_exceptions=True
             )
-        return self._sub_managers[domain]
+            for result in results:
+                if isinstance(result, BaseException):
+                    raise result
+
+            self._sub_managers_ready = True
 
     def _is_domain_throttled(self, domain: str) -> bool:
         """Check if a domain is currently throttled."""
@@ -408,12 +467,19 @@ 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
+    def _fetch_owner(self, request: Request) -> TRequestManager:
+        """Return the manager the request must be given back to, leaving its in-flight record in place.
+
+        The record is dropped by `_clear_fetch_owner` only once the owning manager has accepted the completion, so a
+        completion retried after a transient storage failure still resolves to the same manager.
+        """
+        if (request.unique_key, request.url) in self._in_flight_from_inner:
+            return self._inner
+        return self._sub_managers.get(self._extract_domain(request.url), self._inner)
+
+    def _clear_fetch_owner(self, request: Request) -> None:
+        """Drop the in-flight record of a request whose completion the owning manager has accepted."""
+        self._in_flight_from_inner.discard((request.unique_key, request.url))
 
     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.