WIP

scrapy/core/engine.py

@@ -633,10 +633,13 @@ class ExecutionEngine:
         assert self._slot is not None  # typing
         assert self.spider is not None
 
-        self._slot.add_request(request)
         throttler = self.crawler.throttler
         assert throttler is not None
+        # A throttling-aware scheduler reserves the request (a concurrency slot)
+        # before handing it here, so everything past this point runs inside the
+        # try/finally that releases it, even if add_request() were to raise.
         try:
+            self._slot.add_request(request)
             yield self._acquire_throttling(request)
             result: Response | Request
             if self._downloader_fetch_needs_spider:

scrapy/pqueues.py

@@ -603,6 +603,8 @@ class ThrottlingAwarePriorityQueue:
                 exc_info=True,
                 extra={"spider": getattr(self.crawler, "spider", None)},
             )
+            if self.crawler.stats is not None:
+                self.crawler.stats.inc_value("scheduler/unserializable")
 
     def _select(
         self,

scrapy/throttling.py

@@ -511,8 +511,8 @@ class ThrottlingManager:
         self._default_scope_manager_cls = load_object(
             crawler.settings["THROTTLING_SCOPE_MANAGER"]
         )
-        self._scopes_config: dict[str, dict[str, Any]] = crawler.settings.getdict(
-            "THROTTLING_SCOPES"
+        self._scopes_config: dict[str, dict[str, Any]] = self._merge_download_slots(
+            crawler.settings
         )
         # Ordered by least-recently-used first (see get_scope_manager), so the
         # scope limit can evict the coldest idle scopes (see THROTTLING_SCOPE_LIMIT).
@@ -527,6 +527,35 @@ class ThrottlingManager:
             Request, list[tuple[ThrottlingScopeManagerProtocol, float | None]]
         ] = WeakKeyDictionary()
 
+    @staticmethod
+    def _merge_download_slots(settings: BaseSettings) -> dict[str, dict[str, Any]]:
+        """Return the effective per-scope configuration, merging the deprecated
+        :setting:`DOWNLOAD_SLOTS` setting into :setting:`THROTTLING_SCOPES`.
+
+        Each ``DOWNLOAD_SLOTS`` entry is translated to a throttling scope keyed
+        by the same slot name (the default manager keys domain scopes by
+        ``netloc``, which is what download slots used too): ``concurrency`` and
+        ``delay`` map directly, and the ``randomize_delay`` boolean maps to a
+        ``jitter`` magnitude (the historical ±50%, or none). An explicit
+        ``THROTTLING_SCOPES`` entry for the same scope takes precedence over the
+        translated one. The deprecation warning is emitted by the downloader.
+        """
+        scopes: dict[str, dict[str, Any]] = {
+            scope_id: dict(config)
+            for scope_id, config in settings.getdict("THROTTLING_SCOPES").items()
+        }
+        for slot_id, slot_config in settings.getdict("DOWNLOAD_SLOTS").items():
+            translated: dict[str, Any] = {}
+            if "concurrency" in slot_config:
+                translated["concurrency"] = slot_config["concurrency"]
+            if "delay" in slot_config:
+                translated["delay"] = slot_config["delay"]
+            if "randomize_delay" in slot_config:
+                translated["jitter"] = 0.5 if slot_config["randomize_delay"] else 0.0
+            # An explicit THROTTLING_SCOPES entry wins over the translated one.
+            scopes[slot_id] = {**translated, **scopes.get(slot_id, {})}
+        return scopes
+
     @scope_cache
     async def get_scopes(self, request: Request) -> RequestScopes:
         return self._resolve_scopes_sync(request)