From f25b9a8084659dcccdb17fbbceb06c08ebda9df0 Mon Sep 17 00:00:00 2001
From: Max Bohomolov <moriturus7@gmail.com>
Date: Wed, 19 Aug 2026 21:14:56 +0000
Subject: [PATCH 1/9] open the per-domain sub-queues on first use instead of on
 insert

---
 .../_throttling_request_manager.py            | 105 +++++----
 tests/unit/test_throttling_request_manager.py | 200 +++++++++++++++++-
 2 files changed, 266 insertions(+), 39 deletions(-)

diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py
index d48c49ec92..6e900e2217 100644
--- a/src/crawlee/request_loaders/_throttling_request_manager.py
+++ b/src/crawlee/request_loaders/_throttling_request_manager.py
@@ -35,8 +35,8 @@
 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. 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
@@ -46,9 +46,10 @@ 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
+    The class is generic over the wrapped manager type. On the first call to any of its methods, the
+    `request_manager_opener` callback opens one sub-manager per configured domain, so every sub-manager shares the same
+    `RequestManager` subclass and backing store as `inner`, and requests left over in a persistent store by a previous
+    run are picked up again. The opener must accept `alias`, `storage_client`, and `configuration` keyword arguments
     (as `RequestQueue.open` does) and return the same concrete subclass as `inner`.
 
     ### Usage
@@ -86,9 +87,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 +102,12 @@ 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[str] = set()
+        """Unique keys of requests handed out by `fetch_next_request` from the inner manager. A request whose domain is
+        configured can still live in `inner` if it was added before the domain was listed, and it must be given back to
+        the manager it came from."""
         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 +119,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 +146,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 +171,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 +196,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 +217,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 +227,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 +239,9 @@ async def fetch_next_request(self) -> Request | None:
 
             request = await self._inner.fetch_next_request()
             if request is not None:
+                self._in_flight_from_inner.add(request.unique_key)
                 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,20 +259,23 @@ 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._take_fetch_owner(request)
         result = await manager.reclaim_request(request, forefront=forefront)
         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._take_fetch_owner(request)
         result = await manager.mark_request_as_handled(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())
         )
@@ -271,6 +283,7 @@ async def get_handled_count(self) -> int:
 
     @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())
         )
@@ -278,11 +291,13 @@ async def get_total_count(self) -> int:
 
     @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 +383,24 @@ 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(),
-            )
-        return self._sub_managers[domain]
+    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."""
+        if self._sub_managers_ready:
+            return
+
+        async with self._sub_managers_lock:
+            if self._sub_managers_ready:
+                return
+            await asyncio.gather(*(self._open_sub_manager(domain) for domain in self._domain_states))
+            self._sub_managers_ready = True
 
     def _is_domain_throttled(self, domain: str) -> bool:
         """Check if a domain is currently throttled."""
@@ -408,12 +432,17 @@ def _signal_new_work(self) -> None:
         """Wake `fetch_next_request` if it is sleeping inside a throttle wait."""
         self._new_work_event.set()
 
+    def _take_fetch_owner(self, request: Request) -> TRequestManager:
+        """Return the manager the request must be given back to, clearing its in-flight record."""
+        if request.unique_key in self._in_flight_from_inner:
+            self._in_flight_from_inner.discard(request.unique_key)
+            return self._inner
+        return self._select_manager(request.url)
+
     def _select_manager(self, url: str) -> TRequestManager:
-        """Return the manager that owns the given URL — its sub-manager if one exists, otherwise the inner."""
+        """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
+        return self._sub_managers.get(domain, 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.
diff --git a/tests/unit/test_throttling_request_manager.py b/tests/unit/test_throttling_request_manager.py
index 0451297fff..c580176d2a 100644
--- a/tests/unit/test_throttling_request_manager.py
+++ b/tests/unit/test_throttling_request_manager.py
@@ -4,6 +4,7 @@
 
 import asyncio
 from datetime import datetime, timedelta, timezone
+from pathlib import Path
 from typing import Any
 from unittest.mock import AsyncMock, patch
 
@@ -12,11 +13,13 @@
 from crawlee._request import Request
 from crawlee._service_locator import ServiceLocator
 from crawlee._utils.http import parse_retry_after_header
+from crawlee.configuration import Configuration
 from crawlee.request_loaders._throttling_request_manager import ThrottlingRequestManager
-from crawlee.storage_clients import MemoryStorageClient
+from crawlee.storage_clients import FileSystemStorageClient, MemoryStorageClient
 from crawlee.storages import RequestQueue
 
 THROTTLED_DOMAIN = 'throttled.com'
+SECOND_THROTTLED_DOMAIN = 'slow.com'
 NON_THROTTLED_DOMAIN = 'free.com'
 TEST_DOMAINS = [THROTTLED_DOMAIN]
 
@@ -50,11 +53,38 @@ async def manager(inner_queue: RequestQueue, service_locator: ServiceLocator) ->
     )
 
 
+@pytest.fixture
+def fs_service_locator() -> ServiceLocator:
+    """Create a ServiceLocator backed by the file system, so storages survive a simulated restart."""
+    return ServiceLocator(configuration=Configuration(purge_on_start=False), storage_client=FileSystemStorageClient())
+
+
 def _make_request(url: str) -> Request:
     """Helper to create a Request object."""
     return Request.from_url(url)
 
 
+async def _open_fs_manager(service_locator: ServiceLocator) -> ThrottlingRequestManager[RequestQueue]:
+    """Open a throttling manager over the persistent storage directory, as a fresh process would."""
+    inner_queue = await RequestQueue.open(
+        name='persistent-inner',
+        storage_client=service_locator.get_storage_client(),
+        configuration=service_locator.get_configuration(),
+    )
+    return ThrottlingRequestManager(
+        inner_queue,
+        domains=TEST_DOMAINS,
+        request_manager_opener=RequestQueue.open,
+        service_locator=service_locator,
+    )
+
+
+async def _restart_fs_manager(service_locator: ServiceLocator) -> ThrottlingRequestManager[RequestQueue]:
+    """Simulate a process restart by dropping cached storage instances and reopening the manager."""
+    service_locator.storage_instance_manager.clear_cache()
+    return await _open_fs_manager(service_locator)
+
+
 # ── Request Routing Tests ─────────────────────────────────
 
 
@@ -437,6 +467,59 @@ async def test_mark_request_as_handled_routes_to_inner(
     assert await inner_queue.get_handled_count() == 1
 
 
+async def test_reclaim_returns_inner_request_to_inner(
+    manager: ThrottlingRequestManager[RequestQueue],
+    inner_queue: RequestQueue,
+) -> None:
+    """A request fetched from inner is reclaimed back into inner, even when its domain is configured."""
+    # Simulate a domain that was added to the configured list only after the request had been stored in inner.
+    await inner_queue.add_request(f'https://{THROTTLED_DOMAIN}/page1')
+
+    request = await manager.fetch_next_request()
+    assert request is not None
+
+    await manager.reclaim_request(request)
+
+    assert not await inner_queue.is_empty()
+    assert await manager._sub_managers[THROTTLED_DOMAIN].is_empty()
+
+
+async def test_handled_inner_request_finishes_inner(
+    manager: ThrottlingRequestManager[RequestQueue],
+    inner_queue: RequestQueue,
+) -> None:
+    """A request fetched from inner is marked as handled in inner, so inner can finish."""
+    await inner_queue.add_request(f'https://{THROTTLED_DOMAIN}/page1')
+
+    request = await manager.fetch_next_request()
+    assert request is not None
+
+    await manager.mark_request_as_handled(request)
+
+    assert await inner_queue.is_finished() is True
+    assert await manager.is_finished() is True
+
+
+async def test_purge_forgets_in_flight_inner_requests(
+    manager: ThrottlingRequestManager[RequestQueue],
+    inner_queue: RequestQueue,
+) -> None:
+    """A purge drops the requests still in flight, so their routing must not outlive it."""
+    url = f'https://{THROTTLED_DOMAIN}/page1'
+    await inner_queue.add_request(url)
+    assert await manager.fetch_next_request() is not None
+
+    await manager.purge()
+
+    # The same URL yields the same unique key, so a stale record would send it back to inner.
+    await manager.add_request(url)
+    request = await manager.fetch_next_request()
+    assert request is not None
+    await manager.mark_request_as_handled(request)
+
+    assert await manager._sub_managers[THROTTLED_DOMAIN].is_finished() is True
+
+
 async def test_get_handled_count_aggregates(manager: ThrottlingRequestManager[RequestQueue]) -> None:
     """get_handled_count should sum inner and all sub-managers."""
     throttled_url = f'https://{THROTTLED_DOMAIN}/page1'
@@ -527,6 +610,121 @@ async def test_purge_clears_requests_and_resets_throttle_state(
     assert not manager._is_domain_throttled(THROTTLED_DOMAIN)
 
 
+async def test_sub_managers_opened_for_every_configured_domain(
+    inner_queue: RequestQueue,
+    service_locator: ServiceLocator,
+) -> None:
+    """Every configured domain gets a sub-manager on first use, not during construction."""
+    domains = [THROTTLED_DOMAIN, SECOND_THROTTLED_DOMAIN]
+    manager = ThrottlingRequestManager(
+        inner_queue,
+        domains=domains,
+        request_manager_opener=RequestQueue.open,
+        service_locator=service_locator,
+    )
+
+    assert manager._sub_managers == {}
+
+    await manager.is_empty()
+
+    assert set(manager._sub_managers) == set(domains)
+
+
+async def test_sub_managers_opened_once(
+    inner_queue: RequestQueue,
+    service_locator: ServiceLocator,
+) -> None:
+    """Concurrent first calls open each sub-manager exactly once."""
+    domains = [THROTTLED_DOMAIN, SECOND_THROTTLED_DOMAIN]
+    opener = AsyncMock(side_effect=RequestQueue.open)
+    manager: ThrottlingRequestManager[RequestQueue] = ThrottlingRequestManager(
+        inner_queue,
+        domains=domains,
+        request_manager_opener=opener,
+        service_locator=service_locator,
+    )
+
+    await asyncio.gather(manager.is_empty(), manager.is_finished(), manager.fetch_next_request())
+
+    assert opener.await_count == len(domains)
+
+
+async def test_read_path_after_drop_reopens_sub_managers(
+    manager: ThrottlingRequestManager[RequestQueue],
+) -> None:
+    """After drop(), the next read reopens the sub-managers."""
+    await manager.add_request(f'https://{THROTTLED_DOMAIN}/page1')
+    await manager.drop()
+    assert manager._sub_managers == {}
+
+    assert await manager.is_empty() is True
+    assert set(manager._sub_managers) == set(TEST_DOMAINS)
+
+
+async def test_read_paths_reopen_persisted_sub_queues(fs_service_locator: ServiceLocator) -> None:
+    """A restarted manager sees the requests a previous run left in a persisted sub-queue."""
+    urls = [f'https://{THROTTLED_DOMAIN}/page1', f'https://{THROTTLED_DOMAIN}/page2']
+    manager = await _open_fs_manager(fs_service_locator)
+    await manager.add_requests(urls)
+
+    restarted = await _restart_fs_manager(fs_service_locator)
+
+    assert await restarted.is_finished() is False
+    assert await restarted.is_empty() is False
+    assert await restarted.get_total_count() == 2
+
+    request = await restarted.fetch_next_request()
+    assert request is not None
+    assert request.url in urls
+
+
+async def test_purge_empties_unopened_sub_queues(fs_service_locator: ServiceLocator) -> None:
+    """purge() empties sub-queues left behind by a previous run."""
+    manager = await _open_fs_manager(fs_service_locator)
+    await manager.add_requests([f'https://{THROTTLED_DOMAIN}/page1', f'https://{THROTTLED_DOMAIN}/page2'])
+
+    restarted = await _restart_fs_manager(fs_service_locator)
+    await restarted.purge()
+
+    sub_queue = await RequestQueue.open(
+        alias=f'throttled-{THROTTLED_DOMAIN}',
+        storage_client=fs_service_locator.get_storage_client(),
+        configuration=fs_service_locator.get_configuration(),
+    )
+    assert await sub_queue.get_total_count() == 0
+
+
+async def test_drop_removes_unopened_sub_queues(fs_service_locator: ServiceLocator) -> None:
+    """drop() removes the on-disk sub-queues left behind by a previous run."""
+    manager = await _open_fs_manager(fs_service_locator)
+    await manager.add_request(f'https://{THROTTLED_DOMAIN}/page1')
+
+    sub_queue_path = (
+        Path(fs_service_locator.get_configuration().storage_dir) / 'request_queues' / f'throttled-{THROTTLED_DOMAIN}'
+    )
+    assert sub_queue_path.exists()
+
+    restarted = await _restart_fs_manager(fs_service_locator)
+    await restarted.drop()
+
+    assert not sub_queue_path.exists()
+
+
+async def test_reclaim_routes_to_sub_manager_after_restart(fs_service_locator: ServiceLocator) -> None:
+    """After a restart, a reclaimed request goes back to its sub-manager rather than to inner."""
+    manager = await _open_fs_manager(fs_service_locator)
+    await manager.add_request(f'https://{THROTTLED_DOMAIN}/page1')
+
+    restarted = await _restart_fs_manager(fs_service_locator)
+    request = await restarted.fetch_next_request()
+    assert request is not None
+
+    await restarted.reclaim_request(request)
+
+    assert not await restarted._sub_managers[THROTTLED_DOMAIN].is_empty()
+    assert await restarted.inner.is_empty()
+
+
 # ── Utility Tests ──────────────────────────────────────
 
 

From 3a389f824c861fa10582e6836a044a07383d98f2 Mon Sep 17 00:00:00 2001
From: Vlada Dusek <v.dusek96@gmail.com>
Date: Thu, 20 Aug 2026 14:02:00 +0200
Subject: [PATCH 2/9] fix(throttling-manager): key in-flight inner requests by
 unique key and URL

---
 .../_throttling_request_manager.py            | 18 ++++++++-----
 tests/unit/test_throttling_request_manager.py | 27 +++++++++++++++++++
 2 files changed, 38 insertions(+), 7 deletions(-)

diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py
index 6e900e2217..e02e296e24 100644
--- a/src/crawlee/request_loaders/_throttling_request_manager.py
+++ b/src/crawlee/request_loaders/_throttling_request_manager.py
@@ -104,10 +104,12 @@ def __init__(
         self._sub_managers: dict[str, TRequestManager] = {}
         self._sub_managers_ready = False
         self._sub_managers_lock = asyncio.Lock()
-        self._in_flight_from_inner: set[str] = set()
-        """Unique keys of requests handed out by `fetch_next_request` from the inner manager. A request whose domain is
-        configured can still live in `inner` if it was added before the domain was listed, and it must be given back to
-        the manager it came from."""
+        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."""
         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."""
@@ -239,7 +241,8 @@ async def fetch_next_request(self) -> Request | None:
 
             request = await self._inner.fetch_next_request()
             if request is not None:
-                self._in_flight_from_inner.add(request.unique_key)
+                if self._extract_domain(request.url) in self._domain_states:
+                    self._in_flight_from_inner.add((request.unique_key, request.url))
                 return request
 
             sub_managers_empty = await asyncio.gather(*(sm.is_empty() for sm in self._sub_managers.values()))
@@ -434,8 +437,9 @@ def _signal_new_work(self) -> None:
 
     def _take_fetch_owner(self, request: Request) -> TRequestManager:
         """Return the manager the request must be given back to, clearing its in-flight record."""
-        if request.unique_key in self._in_flight_from_inner:
-            self._in_flight_from_inner.discard(request.unique_key)
+        key = (request.unique_key, request.url)
+        if key in self._in_flight_from_inner:
+            self._in_flight_from_inner.remove(key)
             return self._inner
         return self._select_manager(request.url)
 
diff --git a/tests/unit/test_throttling_request_manager.py b/tests/unit/test_throttling_request_manager.py
index c580176d2a..6e79b8c242 100644
--- a/tests/unit/test_throttling_request_manager.py
+++ b/tests/unit/test_throttling_request_manager.py
@@ -520,6 +520,33 @@ async def test_purge_forgets_in_flight_inner_requests(
     assert await manager._sub_managers[THROTTLED_DOMAIN].is_finished() is True
 
 
+async def test_shared_unique_key_does_not_reroute_sub_manager_request(
+    manager: ThrottlingRequestManager[RequestQueue],
+    inner_queue: RequestQueue,
+) -> None:
+    """A unique key shared with an in-flight inner request must not send a sub-manager request back to inner."""
+    shared_key = 'shared-unique-key'
+    # A leftover from before the domain was configured, sharing an explicit unique key with a freshly added request.
+    await inner_queue.add_request(
+        Request.from_url(f'https://{THROTTLED_DOMAIN}/page1', unique_key=shared_key),
+    )
+    await manager.add_request(Request.from_url(f'https://{THROTTLED_DOMAIN}/page2', unique_key=shared_key))
+
+    # Sub-managers are drained before inner, so the first fetch is the sub-manager's request.
+    sub_request = await manager.fetch_next_request()
+    inner_request = await manager.fetch_next_request()
+    assert sub_request is not None
+    assert inner_request is not None
+    assert sub_request.url == f'https://{THROTTLED_DOMAIN}/page2'
+    assert inner_request.url == f'https://{THROTTLED_DOMAIN}/page1'
+
+    await manager.reclaim_request(sub_request)
+
+    # The reclaim has to land in the sub-manager; inner keeps its own request in flight.
+    assert not await manager._sub_managers[THROTTLED_DOMAIN].is_empty()
+    assert await inner_queue.is_empty() is True
+
+
 async def test_get_handled_count_aggregates(manager: ThrottlingRequestManager[RequestQueue]) -> None:
     """get_handled_count should sum inner and all sub-managers."""
     throttled_url = f'https://{THROTTLED_DOMAIN}/page1'

From 873c127176b06e4bd082936b0426b147d520223d Mon Sep 17 00:00:00 2001
From: Vlada Dusek <v.dusek96@gmail.com>
Date: Thu, 20 Aug 2026 14:02:13 +0200
Subject: [PATCH 3/9] refactor(throttling-manager): inline _select_manager into
 its only call site

---
 src/crawlee/request_loaders/_throttling_request_manager.py | 7 +------
 1 file changed, 1 insertion(+), 6 deletions(-)

diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py
index e02e296e24..027b4675c7 100644
--- a/src/crawlee/request_loaders/_throttling_request_manager.py
+++ b/src/crawlee/request_loaders/_throttling_request_manager.py
@@ -441,12 +441,7 @@ def _take_fetch_owner(self, request: Request) -> TRequestManager:
         if key in self._in_flight_from_inner:
             self._in_flight_from_inner.remove(key)
             return self._inner
-        return self._select_manager(request.url)
-
-    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)
-        return self._sub_managers.get(domain, self._inner)
+        return self._sub_managers.get(self._extract_domain(request.url), 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.

From 2dee3645f756415084f61a3c3deddad0f991d27f Mon Sep 17 00:00:00 2001
From: Vlada Dusek <v.dusek96@gmail.com>
Date: Thu, 20 Aug 2026 14:02:22 +0200
Subject: [PATCH 4/9] test(throttling-manager): cover the default
 purge-on-start path for sub-queues

---
 tests/unit/test_throttling_request_manager.py | 21 +++++++++++++++++--
 1 file changed, 19 insertions(+), 2 deletions(-)

diff --git a/tests/unit/test_throttling_request_manager.py b/tests/unit/test_throttling_request_manager.py
index 6e79b8c242..6c2ac6bf8c 100644
--- a/tests/unit/test_throttling_request_manager.py
+++ b/tests/unit/test_throttling_request_manager.py
@@ -705,7 +705,24 @@ async def test_read_paths_reopen_persisted_sub_queues(fs_service_locator: Servic
     assert request.url in urls
 
 
-async def test_purge_empties_unopened_sub_queues(fs_service_locator: ServiceLocator) -> None:
+async def test_default_purge_on_start_empties_persisted_sub_queues(fs_service_locator: ServiceLocator) -> None:
+    """With the default `purge_on_start`, opening the sub-queues empties what a previous run left in them."""
+    manager = await _open_fs_manager(fs_service_locator)
+    await manager.add_requests([f'https://{THROTTLED_DOMAIN}/page1', f'https://{THROTTLED_DOMAIN}/page2'])
+
+    # Restart under the default configuration, which purges an aliased store as it opens.
+    fs_service_locator.storage_instance_manager.clear_cache()
+    purging_locator = ServiceLocator(configuration=Configuration(), storage_client=FileSystemStorageClient())
+    purged = await _open_fs_manager(purging_locator)
+    assert await purged.is_empty() is True
+
+    # Reopen without purging, so an empty queue proves the requests were deleted rather than just hidden.
+    restarted = await _restart_fs_manager(fs_service_locator)
+    assert await restarted.get_total_count() == 0
+    assert await restarted.is_finished() is True
+
+
+async def test_purge_empties_sub_queues_from_a_previous_run(fs_service_locator: ServiceLocator) -> None:
     """purge() empties sub-queues left behind by a previous run."""
     manager = await _open_fs_manager(fs_service_locator)
     await manager.add_requests([f'https://{THROTTLED_DOMAIN}/page1', f'https://{THROTTLED_DOMAIN}/page2'])
@@ -721,7 +738,7 @@ async def test_purge_empties_unopened_sub_queues(fs_service_locator: ServiceLoca
     assert await sub_queue.get_total_count() == 0
 
 
-async def test_drop_removes_unopened_sub_queues(fs_service_locator: ServiceLocator) -> None:
+async def test_drop_removes_sub_queues_from_a_previous_run(fs_service_locator: ServiceLocator) -> None:
     """drop() removes the on-disk sub-queues left behind by a previous run."""
     manager = await _open_fs_manager(fs_service_locator)
     await manager.add_request(f'https://{THROTTLED_DOMAIN}/page1')

From 986fd269f65fc9652e12d022ac604bb27c2aaa29 Mon Sep 17 00:00:00 2001
From: Vlada Dusek <v.dusek96@gmail.com>
Date: Thu, 20 Aug 2026 14:02:31 +0200
Subject: [PATCH 5/9] docs(throttling-manager): document sub-manager storage
 and purge-on-start behavior

---
 docs/guides/request_throttling.mdx            | 15 +++++++++++++
 .../_throttling_request_manager.py            | 22 ++++++++++++-------
 2 files changed, 29 insertions(+), 8 deletions(-)

diff --git a/docs/guides/request_throttling.mdx b/docs/guides/request_throttling.mdx
index e4c99a6c8d..60dd6298a2 100644
--- a/docs/guides/request_throttling.mdx
+++ b/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.
diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py
index 027b4675c7..d453b65512 100644
--- a/src/crawlee/request_loaders/_throttling_request_manager.py
+++ b/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. 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,11 +48,15 @@ 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. On the first call to any of its methods, the
-    `request_manager_opener` callback opens one sub-manager per configured domain, so every sub-manager shares the same
-    `RequestManager` subclass and backing store as `inner`, and requests left over in a persistent store by a previous
-    run are picked up again. 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. On first use, the `request_manager_opener` callback opens one
+    sub-manager per configured domain, 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`.
+
+    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
 
@@ -132,7 +138,7 @@ 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 stay open, they're just emptied.
+        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()))

From 898e55c556af2df422961164972fc716a86b128b Mon Sep 17 00:00:00 2001
From: Vlada Dusek <v.dusek96@gmail.com>
Date: Thu, 20 Aug 2026 14:21:43 +0200
Subject: [PATCH 6/9] fix(throttling-manager): keep the in-flight record until
 the completion is accepted

---
 .../_throttling_request_manager.py            | 22 +++++++++++-----
 tests/unit/test_throttling_request_manager.py | 26 +++++++++++++++++++
 2 files changed, 41 insertions(+), 7 deletions(-)

diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py
index d453b65512..1f62dbcc41 100644
--- a/src/crawlee/request_loaders/_throttling_request_manager.py
+++ b/src/crawlee/request_loaders/_throttling_request_manager.py
@@ -269,16 +269,18 @@ async def fetch_next_request(self) -> Request | None:
     @override
     async def reclaim_request(self, request: Request, *, forefront: bool = False) -> ProcessedRequest | None:
         await self._ensure_sub_managers()
-        manager = self._take_fetch_owner(request)
+        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:
         await self._ensure_sub_managers()
-        manager = self._take_fetch_owner(request)
+        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
 
@@ -441,14 +443,20 @@ def _signal_new_work(self) -> None:
         """Wake `fetch_next_request` if it is sleeping inside a throttle wait."""
         self._new_work_event.set()
 
-    def _take_fetch_owner(self, request: Request) -> TRequestManager:
-        """Return the manager the request must be given back to, clearing its in-flight record."""
-        key = (request.unique_key, request.url)
-        if key in self._in_flight_from_inner:
-            self._in_flight_from_inner.remove(key)
+    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.
 
diff --git a/tests/unit/test_throttling_request_manager.py b/tests/unit/test_throttling_request_manager.py
index 6c2ac6bf8c..9acbc95eef 100644
--- a/tests/unit/test_throttling_request_manager.py
+++ b/tests/unit/test_throttling_request_manager.py
@@ -520,6 +520,32 @@ async def test_purge_forgets_in_flight_inner_requests(
     assert await manager._sub_managers[THROTTLED_DOMAIN].is_finished() is True
 
 
+async def test_failed_completion_keeps_its_inner_routing_for_the_retry(
+    manager: ThrottlingRequestManager[RequestQueue],
+    inner_queue: RequestQueue,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A completion that fails keeps its inner routing, so a retried completion is not diverted to the sub-manager."""
+    await inner_queue.add_request(f'https://{THROTTLED_DOMAIN}/page1')
+    request = await manager.fetch_next_request()
+    assert request is not None
+
+    # `BasicCrawler` retries `mark_request_as_handled`, so a failed attempt must not consume the routing record.
+    monkeypatch.setattr(
+        inner_queue,
+        'mark_request_as_handled',
+        AsyncMock(side_effect=RuntimeError('transient storage failure')),
+    )
+    with pytest.raises(RuntimeError, match='transient storage failure'):
+        await manager.mark_request_as_handled(request)
+
+    monkeypatch.undo()
+    await manager.mark_request_as_handled(request)
+
+    assert await inner_queue.is_finished() is True
+    assert await manager.is_finished() is True
+
+
 async def test_shared_unique_key_does_not_reroute_sub_manager_request(
     manager: ThrottlingRequestManager[RequestQueue],
     inner_queue: RequestQueue,

From 18825311ce682ad0aeea81505314b59578cf9bff Mon Sep 17 00:00:00 2001
From: Vlada Dusek <v.dusek96@gmail.com>
Date: Thu, 20 Aug 2026 14:21:52 +0200
Subject: [PATCH 7/9] fix(throttling-manager): keep sub-managers that opened
 before a sibling failed

---
 .../_throttling_request_manager.py            | 19 +++++++++--
 tests/unit/test_throttling_request_manager.py | 34 +++++++++++++++++++
 2 files changed, 51 insertions(+), 2 deletions(-)

diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py
index 1f62dbcc41..af08215467 100644
--- a/src/crawlee/request_loaders/_throttling_request_manager.py
+++ b/src/crawlee/request_loaders/_throttling_request_manager.py
@@ -403,14 +403,29 @@ async def _open_sub_manager(self, domain: str) -> None:
         )
 
     async def _ensure_sub_managers(self) -> None:
-        """Open a sub-manager for every configured domain, once."""
+        """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
-            await asyncio.gather(*(self._open_sub_manager(domain) for domain in self._domain_states))
+
+            # 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
+            )
+            for result in results:
+                if isinstance(result, BaseException):
+                    raise result
+
             self._sub_managers_ready = True
 
     def _is_domain_throttled(self, domain: str) -> bool:
diff --git a/tests/unit/test_throttling_request_manager.py b/tests/unit/test_throttling_request_manager.py
index 9acbc95eef..21eccbcf4a 100644
--- a/tests/unit/test_throttling_request_manager.py
+++ b/tests/unit/test_throttling_request_manager.py
@@ -702,6 +702,40 @@ async def test_sub_managers_opened_once(
     assert opener.await_count == len(domains)
 
 
+async def test_failed_open_keeps_successful_sub_managers(
+    inner_queue: RequestQueue,
+    service_locator: ServiceLocator,
+) -> None:
+    """A failing opener leaves the sub-managers that did open in place, so a retry opens only what is missing."""
+    domains = [THROTTLED_DOMAIN, SECOND_THROTTLED_DOMAIN]
+    failing_alias = f'throttled-{SECOND_THROTTLED_DOMAIN}'
+    pending_failures = {failing_alias}
+
+    async def open_once_failing(*, alias: str, **kwargs: Any) -> RequestQueue:
+        if alias in pending_failures:
+            pending_failures.discard(alias)
+            raise RuntimeError('storage unavailable')
+        return await RequestQueue.open(alias=alias, **kwargs)
+
+    opener = AsyncMock(side_effect=open_once_failing)
+    manager: ThrottlingRequestManager[RequestQueue] = ThrottlingRequestManager(
+        inner_queue,
+        domains=domains,
+        request_manager_opener=opener,
+        service_locator=service_locator,
+    )
+
+    with pytest.raises(RuntimeError, match='storage unavailable'):
+        await manager.is_empty()
+
+    assert await manager.is_empty() is True
+
+    # The domain that opened before its sibling failed is reused, not opened a second time.
+    opened_aliases = [call.kwargs['alias'] for call in opener.await_args_list]
+    assert opened_aliases.count(f'throttled-{THROTTLED_DOMAIN}') == 1
+    assert opened_aliases.count(failing_alias) == 2
+
+
 async def test_read_path_after_drop_reopens_sub_managers(
     manager: ThrottlingRequestManager[RequestQueue],
 ) -> None:

From 3fff872db05856da93911975378be947d8291f75 Mon Sep 17 00:00:00 2001
From: Vlada Dusek <v.dusek96@gmail.com>
Date: Thu, 20 Aug 2026 14:21:53 +0200
Subject: [PATCH 8/9] docs(throttling-manager): clarify when sub-managers open
 and how in-flight records are keyed

---
 .../_throttling_request_manager.py              | 17 +++++++++++++----
 1 file changed, 13 insertions(+), 4 deletions(-)

diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py
index af08215467..d7ad0a5a14 100644
--- a/src/crawlee/request_loaders/_throttling_request_manager.py
+++ b/src/crawlee/request_loaders/_throttling_request_manager.py
@@ -48,9 +48,11 @@ 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. On first use, the `request_manager_opener` callback opens one
-    sub-manager per configured domain, 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
+    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
@@ -115,7 +117,14 @@ def __init__(
         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."""
+        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."""

From 255ed412983dff8557ac12137751622dc7d479b3 Mon Sep 17 00:00:00 2001
From: Vlada Dusek <v.dusek96@gmail.com>
Date: Thu, 20 Aug 2026 14:31:49 +0200
Subject: [PATCH 9/9] docs(throttling-manager): tighten the comments and
 docstrings

---
 .../_throttling_request_manager.py            | 57 +++++++------------
 tests/unit/test_throttling_request_manager.py |  8 +--
 2 files changed, 24 insertions(+), 41 deletions(-)

diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py
index d7ad0a5a14..c7b5389327 100644
--- a/src/crawlee/request_loaders/_throttling_request_manager.py
+++ b/src/crawlee/request_loaders/_throttling_request_manager.py
@@ -35,10 +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. 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.
+    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.
 
     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
@@ -48,17 +47,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 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`.
+    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`.
 
-    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.
+    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
 
@@ -113,18 +109,11 @@ def __init__(
         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."""
+        """`(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."""
         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."""
@@ -412,11 +401,7 @@ async def _open_sub_manager(self, domain: str) -> None:
         )
 
     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.
-        """
+        """Open a sub-manager for every configured domain, once; a retry opens only what is still missing."""
         if self._sub_managers_ready:
             return
 
@@ -424,9 +409,8 @@ async def _ensure_sub_managers(self) -> None:
             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.
+            # 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
@@ -470,15 +454,14 @@ def _signal_new_work(self) -> None:
     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.
+        `_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 the owning manager has accepted."""
+        """Drop the in-flight record of a request whose completion was accepted."""
         self._in_flight_from_inner.discard((request.unique_key, request.url))
 
     async def _wait_for_new_work_or_timeout(self, timeout: float) -> None:
diff --git a/tests/unit/test_throttling_request_manager.py b/tests/unit/test_throttling_request_manager.py
index 21eccbcf4a..162f5b3e6e 100644
--- a/tests/unit/test_throttling_request_manager.py
+++ b/tests/unit/test_throttling_request_manager.py
@@ -472,7 +472,7 @@ async def test_reclaim_returns_inner_request_to_inner(
     inner_queue: RequestQueue,
 ) -> None:
     """A request fetched from inner is reclaimed back into inner, even when its domain is configured."""
-    # Simulate a domain that was added to the configured list only after the request had been stored in inner.
+    # A domain configured only after the request had already been stored in inner.
     await inner_queue.add_request(f'https://{THROTTLED_DOMAIN}/page1')
 
     request = await manager.fetch_next_request()
@@ -530,7 +530,7 @@ async def test_failed_completion_keeps_its_inner_routing_for_the_retry(
     request = await manager.fetch_next_request()
     assert request is not None
 
-    # `BasicCrawler` retries `mark_request_as_handled`, so a failed attempt must not consume the routing record.
+    # `BasicCrawler` retries this, so a failed attempt must not consume the routing record.
     monkeypatch.setattr(
         inner_queue,
         'mark_request_as_handled',
@@ -552,7 +552,7 @@ async def test_shared_unique_key_does_not_reroute_sub_manager_request(
 ) -> None:
     """A unique key shared with an in-flight inner request must not send a sub-manager request back to inner."""
     shared_key = 'shared-unique-key'
-    # A leftover from before the domain was configured, sharing an explicit unique key with a freshly added request.
+    # A leftover from before the domain was configured, sharing its explicit unique key with a fresh request.
     await inner_queue.add_request(
         Request.from_url(f'https://{THROTTLED_DOMAIN}/page1', unique_key=shared_key),
     )
@@ -776,7 +776,7 @@ async def test_default_purge_on_start_empties_persisted_sub_queues(fs_service_lo
     purged = await _open_fs_manager(purging_locator)
     assert await purged.is_empty() is True
 
-    # Reopen without purging, so an empty queue proves the requests were deleted rather than just hidden.
+    # Reopen without purging: an empty queue proves the requests were deleted, not just hidden.
     restarted = await _restart_fs_manager(fs_service_locator)
     assert await restarted.get_total_count() == 0
     assert await restarted.is_finished() is True