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

docs/guides/request_throttling.mdx

@@ -32,7 +32,7 @@ To use request throttling, create a <ApiLink to="class/ThrottlingRequestManager"
 
 ## How it works
 
-1. **Insertion-time routing**: When you add requests via `add_request` or `add_requests`, each request is checked against the configured domain list. Matching requests go directly into a per-domain sub-manager; all others go to the inner manager. This eliminates request duplication entirely.
+1. **Insertion-time routing**: When you add requests via `add_request` or `add_requests`, each request is checked against the configured domain list. Matching requests go directly into a per-domain sub-manager; all others go to the inner manager. Each request added this way lives in exactly one store, so it is deduplicated there.
 
 2. **429 backoff**: When the crawler detects an HTTP 429 response, the `ThrottlingRequestManager` records an exponential backoff delay for that domain (starting at 2s, doubling up to 60s). If the response includes a `Retry-After` header, that value takes priority.
 
@@ -42,6 +42,21 @@ To use request throttling, create a <ApiLink to="class/ThrottlingRequestManager"
 
 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.
 
+## 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,9 @@
 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, so each request lives in exactly
+    one store and is deduplicated there. A request that reached `inner` before its domain was configured stays and is
+    completed there, without the domain's delay.
 
     `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
@@ -48,10 +49,14 @@ 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 opens one sub-manager per
+    configured domain through `request_manager_opener`, so all of them share the subclass and backing store of `inner`;
+    the synchronous delay methods 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`.
+
+    Requests a previous run left in a persistent store become visible again at open. The default `purge_on_start=True`
+    empties them; `purge_on_start=False` resumes them. Named stores are exempt from that purge and aliased ones are not,
+    so a named `inner` keeps its requests while the per-domain stores are emptied.
 
     ### Usage
 
@@ -91,9 +96,9 @@ def __init__(
                 per-domain sub-managers. Matching is exact but spelling-insensitive: casing, punycode versus Unicode,
                 and a trailing root dot are all normalized away. 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.
@@ -111,6 +116,14 @@ def __init__(
         domain_keys = [self._parse_configured_domain(entry) for d in domains if (entry := d.strip())]
         self._domain_states: dict[str, _DomainState] = {key: _DomainState(domain=key) for key in domain_keys}
         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 that `fetch_next_request` took from `inner`, where
+        they live if they were added before their domain was listed, and where they must be completed. The URL is part
+        of the key because an explicit `unique_key` is only unique per store. Identical pairs held by `inner` and by a
+        sub-manager are indistinguishable, so their completions can cross; both stores hold the key, so the cost is a
+        duplicate crawl and a retry without the domain's delay."""
 
     @property
     def inner(self) -> TRequestManager:
@@ -119,18 +132,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
@@ -142,12 +159,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)
-            return await sm.add_request(request, forefront=forefront)
+            return await self._sub_managers[domain].add_request(request, forefront=forefront)
 
         return await self._inner.add_request(request, forefront=forefront)
 
@@ -163,6 +181,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]] = {}
 
@@ -186,8 +206,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,
@@ -209,35 +228,47 @@ async def fetch_next_request(self) -> Request | None:
             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.
         """
+        await self._ensure_sub_managers()
+
         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
 
-        return await self._inner.fetch_next_request()
+        request = await self._inner.fetch_next_request()
+        if request is not None and self._extract_domain(request.url) in self._domain_states:
+            self._in_flight_from_inner.add((request.unique_key, request.url))
+        return request
 
     @override
     async def reclaim_request(self, request: Request, *, forefront: bool = False) -> ProcessedRequest | None:
-        manager = self._select_manager(request.url)
-        return await manager.reclaim_request(request, forefront=forefront)
+        await self._ensure_sub_managers()
+        manager = self._fetch_owner(request)
+        result = await manager.reclaim_request(request, forefront=forefront)
+        self._clear_fetch_owner(request)
+        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())
         )
@@ -250,13 +281,15 @@ async def is_empty(self) -> bool:
         Requests queued for a domain in a cooldown do not count. They still count towards `is_finished`, so the crawl
         waits for them.
         """
+        await self._ensure_sub_managers()
         results = await asyncio.gather(
             self._inner.is_empty(), *(self._sub_managers[d].is_empty() for d in self._fetchable_domains())
         )
         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())
         )
@@ -378,15 +411,34 @@ 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; a retry opens only what is still missing."""
+        if self._sub_managers_ready:
+            return
+
+        async with self._sub_managers_lock:
+            if self._sub_managers_ready:
+                return
+
+            # All attempts must settle before the lock is released: openers left running would write into
+            # `_sub_managers` after a retry has already replaced that domain's manager.
+            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."""
@@ -398,13 +450,7 @@ def _is_domain_throttled(self, domain: str) -> bool:
     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 = [domain for domain, state in self._domain_states.items() if now >= state.throttled_until]
         available.sort(key=lambda domain: self._domain_states[domain].throttled_until)
         return available
 
@@ -417,12 +463,18 @@ 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 _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.
+
+        `_clear_fetch_owner` drops the record only once the completion is accepted, so a retry resolves the same way.
+        """
+        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 was accepted."""
+        self._in_flight_from_inner.discard((request.unique_key, request.url))
 
 
 class _RequestManagerOpener(Protocol[TRequestManager]):