Merge 060f1b3ede into ad43bf0c56

docs/news.rst

@@ -5835,9 +5835,9 @@ New features
 *   A new ``DNS_RESOLVER`` setting allows enabling IPv6 support
     (:gh:`1031`, :gh:`4227`)
 
-*   A new :setting:`SCRAPER_SLOT_MAX_ACTIVE_SIZE` setting allows configuring
-    the existing soft limit that pauses request downloads when the total
-    response data being processed is too high (:gh:`1410`, :gh:`3551`)
+*   A new ``SCRAPER_SLOT_MAX_ACTIVE_SIZE`` setting allows configuring the
+    existing soft limit that pauses request downloads when the total response
+    data being processed is too high (:gh:`1410`, :gh:`3551`)
 
 *   A new :setting:`TWISTED_REACTOR` setting allows customizing the
     :mod:`~twisted.internet.reactor` that Scrapy uses, allowing to

docs/topics/practices.rst

@@ -525,6 +525,35 @@ run:
 Because :setting:`SPIDER_MODULES` is a list setting, you can include multiple
 modules by separating them with commas.
 
+.. _crawl-bottlenecks:
+
+Identifying crawl bottlenecks
+=============================
+
+Scrapy exposes ``request_backout_seconds`` stats that show how long request
+scheduling was paused during a crawl, and why:
+
+- ``request_backout_seconds/total``: total time paused for any reason
+- ``request_backout_seconds/concurrency``: time paused because
+  :setting:`CONCURRENT_REQUESTS` was reached
+- ``request_backout_seconds/response_max_active_size``: time paused because
+  :setting:`RESPONSE_MAX_ACTIVE_SIZE` was reached
+
+For example, after a crawl you might see::
+
+    2025-01-01 00:00:00 [scrapy.statscollectors] INFO: Dumping Scrapy stats:
+    {'request_backout_seconds/concurrency': 12.5,
+     'request_backout_seconds/response_max_active_size': 45.2,
+     'request_backout_seconds/total': 57.7,
+     ...}
+
+In this case, the spider spent about 45 seconds paused due to large responses
+in memory. You could:
+
+- Increase :setting:`RESPONSE_MAX_ACTIVE_SIZE` if your machine has enough RAM.
+- Check that your code doesn't hold strong references to
+  :class:`~scrapy.http.Response` objects longer than necessary.
+
 .. _bans:
 
 Avoiding getting banned

docs/topics/settings.rst

@@ -1865,6 +1865,56 @@ Adjust redirect request priority relative to original request:
 - **a positive priority adjust (default) means higher priority.**
 - a negative priority adjust means lower priority.
 
+.. setting:: RESPONSE_MAX_ACTIVE_SIZE
+
+RESPONSE_MAX_ACTIVE_SIZE
+------------------------
+
+Default: ``5_000_000``
+
+Soft limit (in bytes) on the total size of responses being kept in memory.
+
+This counts both the size of response bodies that have passed through
+:ref:`downloader middlewares <topics-downloader-middleware>` and remain in
+memory, and the :setting:`rough size <RESPONSE_ROUGH_SIZE>` of requests
+currently being downloaded.
+
+While the total is above this value, Scrapy pauses sending new requests to the
+downloader. A higher value improves crawl speed at the cost of memory usage.
+``0`` disables the limit.
+
+Scrapy logs an info-level message the first time the limit is reached. To see
+how long request processing was paused because of it over a whole crawl, check
+the ``request_backout_seconds/response_max_active_size`` stat.
+
+.. caution::
+
+    Responses that your code keeps a strong reference to, e.g. in the
+    :attr:`.Request.meta` of a scheduled request or in a component
+    attribute, count toward this limit until that reference is gone, so
+    accumulating them can pause a crawl indefinitely. To find out, run
+    ``prefs()`` on the :ref:`telnet console <topics-telnetconsole>` and see
+    whether the count of live :class:`~scrapy.http.Response` objects keeps
+    growing; see :ref:`topics-leaks`.
+
+.. versionadded:: VERSION
+
+.. setting:: RESPONSE_ROUGH_SIZE
+
+RESPONSE_ROUGH_SIZE
+-------------------
+
+Default: ``None``
+
+Estimated size (in bytes) to count toward :setting:`RESPONSE_MAX_ACTIVE_SIZE`
+for each request being downloaded, whose actual response size is not known yet.
+
+``None`` means a quarter of :setting:`RESPONSE_MAX_ACTIVE_SIZE` split among
+:setting:`CONCURRENT_REQUESTS` requests, so that requests being downloaded
+cannot use up the whole limit on their own. ``0`` counts only responses.
+
+.. versionadded:: VERSION
+
 .. setting:: ROBOTSTXT_OBEY
 
 ROBOTSTXT_OBEY
@@ -2017,19 +2067,6 @@ For available choices, see :setting:`SCHEDULER_MEMORY_QUEUE`.
     :start-after: queue-common-starts
     :end-before: queue-common-ends
 
-
-.. setting:: SCRAPER_SLOT_MAX_ACTIVE_SIZE
-
-SCRAPER_SLOT_MAX_ACTIVE_SIZE
-----------------------------
-
-Default: ``5_000_000``
-
-Soft limit (in bytes) for response data being processed.
-
-While the sum of the sizes of all responses being processed is above this value,
-Scrapy does not process new requests.
-
 .. setting:: SPIDER_CONTRACTS
 
 SPIDER_CONTRACTS

docs/topics/telnetconsole.rst

@@ -125,9 +125,8 @@ engine status::
     len(engine._slot.scheduler.mqs)                 : 92
     len(engine.scraper.slot.queue)                  : 0
     len(engine.scraper.slot.active)                 : 0
-    engine.scraper.slot.active_size                 : 0
+    engine.downloader.middleware._total_active_size : 1310720
     engine.scraper.slot.itemproc_size               : 0
-    engine.scraper.slot.needs_backout()             : False
 
 
 Pause, resume and stop the Scrapy engine

scrapy/core/downloader/__init__.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
+import gc
 import random
 from collections import deque
 from dataclasses import dataclass, field
 from datetime import datetime
+from logging import getLogger
 from time import monotonic
 from typing import TYPE_CHECKING, Any
 
@@ -40,6 +42,8 @@ if TYPE_CHECKING:
     from scrapy.settings import BaseSettings
     from scrapy.signalmanager import SignalManager
 
+logger = getLogger(__name__)
+
 
 @dataclass(slots=True, eq=False)
 class Slot:
@@ -83,6 +87,7 @@ class Slot:
 class Downloader:
     DOWNLOAD_SLOT = "download_slot"
     _SLOT_GC_INTERVAL: float = 60.0  # seconds
+    _GC_INTERVAL: float = 1.0  # seconds
 
     def __init__(self, crawler: Crawler):
         self.crawler: Crawler = crawler
@@ -107,6 +112,12 @@ class Downloader:
         self.per_slot_settings: dict[str, dict[str, Any]] = self.settings.getdict(
             "DOWNLOAD_SLOTS"
         )
+        self._stats = crawler.stats
+        # (reason, start_time): current backout reason and when it began, or
+        # (None, None) if not backing out.
+        self._last_backout: tuple[str | None, float | None] = (None, None)
+        self._max_active_size_warned = False
+        self._last_gc: float = 0
 
     @inlineCallbacks
     @_warn_spider_arg
@@ -114,6 +125,7 @@ class Downloader:
         self, request: Request, spider: Spider | None = None
     ) -> Generator[Deferred[Any], Any, Response | Request]:
         self.active.add(request)
+        self.middleware._count_rough_size(request)
         try:
             result: Response | Request = yield (
                 deferred_from_coro(
@@ -123,10 +135,51 @@ class Downloader:
             return result
         finally:
             self.active.remove(request)
+            self.middleware._discount_rough_size(request)
+
+    def _record_backout(self, reason: str | None) -> None:
+        last_reason, last_reason_start_time = self._last_backout
+        if last_reason == reason:
+            return
+        current_time = monotonic()
+        if last_reason is not None and self._stats is not None:
+            assert last_reason_start_time is not None
+            last_reason_seconds = current_time - last_reason_start_time
+            self._stats.inc_value("request_backout_seconds/total", last_reason_seconds)
+            self._stats.inc_value(
+                f"request_backout_seconds/{last_reason}", last_reason_seconds
+            )
+        self._last_backout = (reason, current_time)
 
     def needs_backout(self) -> bool:
         # A total concurrency of 0 means no limit.
-        return 0 < self.total_concurrency <= len(self.active)
+        if 0 < self.total_concurrency <= len(self.active):
+            self._record_backout("concurrency")
+            return True
+        max_active_size = self.middleware._max_active_size
+        if max_active_size and self.middleware._total_active_size >= max_active_size:
+            if not self._max_active_size_warned:
+                self._max_active_size_warned = True
+                logger.info(
+                    f"Pausing request processing: the active response size "
+                    f"({self.middleware._total_active_size} B) has reached "
+                    f"RESPONSE_MAX_ACTIVE_SIZE ({max_active_size} B). See "
+                    f"https://docs.scrapy.org/en/latest/topics/settings.html#response-max-active-size "
+                    f"and the request_backout_seconds/response_max_active_size "
+                    f"stat. This message is only logged once."
+                )
+            self._record_backout("response_max_active_size")
+            # Responses are only freed once nothing references them, which for
+            # reference cycles, and for every response on PyPy, requires a
+            # garbage collection. A full collection is expensive, and this runs
+            # once per request while paused, hence the interval.
+            current_time = monotonic()
+            if current_time - self._last_gc >= self._GC_INTERVAL:
+                self._last_gc = current_time
+                gc.collect()
+            return True
+        self._record_backout(None)
+        return False
 
     @_warn_spider_arg
     def _get_slot(
@@ -248,6 +301,7 @@ class Downloader:
         self._stop_slot_gc()
         for slot in self.slots.values():
             slot.close()
+        self._record_backout(None)
 
     def _slot_gc(self, age: float = 60) -> None:
         mintime = monotonic() - age

scrapy/core/downloader/middleware.py

@@ -7,8 +7,9 @@ See documentation in docs/topics/downloader-middleware.rst
 from __future__ import annotations
 
 import warnings
-from functools import wraps
+from functools import partial, wraps
 from typing import TYPE_CHECKING, Any
+from weakref import WeakSet, finalize
 
 from scrapy.exceptions import ScrapyDeprecationWarning, _InvalidOutput
 from scrapy.http import Request, Response
@@ -31,9 +32,61 @@ if TYPE_CHECKING:
     from scrapy.settings import BaseSettings
 
 
+def _get_max_active_size(settings: BaseSettings) -> int:
+    deprecated_priority = settings.getpriority("SCRAPER_SLOT_MAX_ACTIVE_SIZE")
+    priority = settings.getpriority("RESPONSE_MAX_ACTIVE_SIZE")
+    assert deprecated_priority is not None
+    assert priority is not None
+    if deprecated_priority <= 0:
+        return settings.getint("RESPONSE_MAX_ACTIVE_SIZE")
+    if priority >= deprecated_priority:
+        warnings.warn(
+            "The SCRAPER_SLOT_MAX_ACTIVE_SIZE setting is deprecated and is "
+            "being ignored because RESPONSE_MAX_ACTIVE_SIZE is set with an "
+            "equal or higher priority. Remove SCRAPER_SLOT_MAX_ACTIVE_SIZE "
+            "from your settings.",
+            ScrapyDeprecationWarning,
+            stacklevel=2,
+        )
+        return settings.getint("RESPONSE_MAX_ACTIVE_SIZE")
+    warnings.warn(
+        "The SCRAPER_SLOT_MAX_ACTIVE_SIZE setting is deprecated, use "
+        "RESPONSE_MAX_ACTIVE_SIZE instead.",
+        ScrapyDeprecationWarning,
+        stacklevel=2,
+    )
+    return settings.getint("SCRAPER_SLOT_MAX_ACTIVE_SIZE")
+
+
+def _get_response_rough_size(settings: BaseSettings, max_active_size: int) -> int:
+    if settings.get("RESPONSE_ROUGH_SIZE") is not None:
+        return settings.getint("RESPONSE_ROUGH_SIZE")
+    concurrency = settings.getint("CONCURRENT_REQUESTS")
+    if not concurrency:
+        # Unlimited concurrency: there is no bound on the number of in-flight
+        # requests to spread a share of the limit over.
+        return 0
+    # Requests being downloaded, whose response size is unknown, may take up to
+    # a quarter of the limit; the rest is for responses already in memory.
+    return max_active_size // (4 * concurrency)
+
+
 class DownloaderMiddlewareManager(MiddlewareManager):
     component_name = "downloader middleware"
 
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        assert self.crawler is not None
+        settings = self.crawler.settings
+        self._max_active_size: int = _get_max_active_size(settings)
+        self._response_rough_size: int = _get_response_rough_size(
+            settings, self._max_active_size
+        )
+        self._response_active_size = 0
+        self._tracked_responses: WeakSet[Response] = WeakSet()
+        self._rough_active_size = 0
+        self._rough_sizes: dict[Request, int] = {}
+
     @classmethod
     def _get_mwlist_from_settings(cls, settings: BaseSettings) -> list[Any]:
         return build_component_list(
@@ -51,6 +104,36 @@ class DownloaderMiddlewareManager(MiddlewareManager):
             self.methods["process_exception"].appendleft(mw.process_exception)
             self._check_mw_method_spider_arg(mw.process_exception)
 
+    @property
+    def _total_active_size(self) -> int:
+        return self._response_active_size + self._rough_active_size
+
+    def _count_rough_size(self, request: Request) -> None:
+        # Counting the same request twice, which concurrent downloads of the
+        # same request object would do, would leak its rough size, since only
+        # one discount call can find it.
+        if request in self._rough_sizes:
+            return
+        self._rough_sizes[request] = self._response_rough_size
+        self._rough_active_size += self._response_rough_size
+
+    def _discount_rough_size(self, request: Request) -> None:
+        self._rough_active_size -= self._rough_sizes.pop(request, 0)
+
+    def _count_response_size(self, response: Response, request: Request) -> None:
+        # The actual response size replaces the rough size estimate of its
+        # request.
+        self._discount_rough_size(request)
+        if response in self._tracked_responses:
+            return
+        self._tracked_responses.add(response)
+        size = len(response.body)
+        self._response_active_size += size
+        finalize(response, partial(self._discount_response_size, size))
+
+    def _discount_response_size(self, size: int) -> None:
+        self._response_active_size -= size
+
     def download(
         self,
         download_func: Callable[[Request, Spider], Deferred[Response]],
@@ -110,8 +193,12 @@ class DownloaderMiddlewareManager(MiddlewareManager):
                     f"Request, got {response.__class__.__name__}"
                 )
             if response:
+                if isinstance(response, Response):
+                    self._count_response_size(response, request)
                 return response
-        return await download_func(request)
+        response = await download_func(request)
+        self._count_response_size(response, request)
+        return response
 
     async def _process_response(
         self, response: Response | Request, request: Request
@@ -135,10 +222,11 @@ class DownloaderMiddlewareManager(MiddlewareManager):
                 )
             if isinstance(response, Request):
                 return response
+            self._count_response_size(response, request)
         return response
 
     async def _process_exception(
-        self, exception: Exception, request: Request | Response
+        self, exception: Exception, request: Request
     ) -> Response | Request:
         for method in self.methods["process_exception"]:
             assert method is not None
@@ -152,5 +240,7 @@ class DownloaderMiddlewareManager(MiddlewareManager):
                     f"Request, got {type(response)}"
                 )
             if response:
+                if isinstance(response, Response):
+                    self._count_response_size(response, request)
                 return response
         raise exception

scrapy/core/engine.py

@@ -366,7 +366,6 @@ class ExecutionEngine:
             or not self._slot
             or bool(self._slot.closing)
             or self.downloader.needs_backout()
-            or self.scraper.slot.needs_backout()
         )
 
     def _remove_request(self, _: Any, request: Request) -> None:

scrapy/core/scraper.py

@@ -59,18 +59,118 @@ _T = TypeVar("_T")
 QueueTuple: TypeAlias = tuple[Response | Failure, Request, Deferred[None]]
 
 
-class Slot:
+_UNSET = object()
+
+
+def _warn_min_response_size() -> None:
+    warnings.warn(
+        "scrapy.core.scraper.Slot.MIN_RESPONSE_SIZE is deprecated.",
+        ScrapyDeprecationWarning,
+        stacklevel=3,
+    )
+
+
+class _MinResponseSize:
+    """Deprecated alias of ``_MIN_RESPONSE_SIZE``, readable and writable both on
+    the class and on its instances."""
+
+    def __get__(self, instance: Slot | None, owner: type[Slot] | None = None) -> int:
+        _warn_min_response_size()
+        target = instance if instance is not None else owner
+        assert target is not None
+        return target._MIN_RESPONSE_SIZE
+
+    def __set__(self, instance: Slot, value: int) -> None:
+        _warn_min_response_size()
+        instance._MIN_RESPONSE_SIZE = value
+
+
+class _SlotMeta(type):
+    # Class-level assignment bypasses the _MinResponseSize descriptor.
+    def __setattr__(cls, name: str, value: Any) -> None:
+        if name == "MIN_RESPONSE_SIZE":
+            _warn_min_response_size()
+            name = "_MIN_RESPONSE_SIZE"
+        super().__setattr__(name, value)
+
+
+class Slot(metaclass=_SlotMeta):
     """Scraper slot (one per running spider)"""
 
-    MIN_RESPONSE_SIZE = 1024
+    _MIN_RESPONSE_SIZE = 1024
+    # Any so that mypy allows class-level assignment, which _SlotMeta redirects
+    # to _MIN_RESPONSE_SIZE.
+    MIN_RESPONSE_SIZE: Any = _MinResponseSize()
 
-    def __init__(self, max_active_size: int = 5000000):
-        self.max_active_size: int = max_active_size
+    def __init__(self, max_active_size: Any = _UNSET):
+        if max_active_size is _UNSET:
+            max_active_size = 5_000_000
+        else:
+            warnings.warn(
+                (
+                    "The max_active_size parameter of "
+                    "scrapy.core.scraper.Slot is deprecated. Use the "
+                    "RESPONSE_MAX_ACTIVE_SIZE setting instead."
+                ),
+                ScrapyDeprecationWarning,
+                stacklevel=2,
+            )
+        self._max_active_size: int = max_active_size
         self.queue: deque[QueueTuple] = deque()
         self.active: set[Request] = set()
-        self.active_size: int = 0
         self.itemproc_size: int = 0  # just for scrapy.utils.engine.get_engine_status()
         self.closing: Deferred[Spider] | None = None
+        self._active_size: int = 0
+
+    @property
+    def active_size(self) -> int:
+        warnings.warn(
+            (
+                "scrapy.core.scraper.Slot.active_size is deprecated. The size "
+                "of responses in memory is now tracked by the downloader, and "
+                "no longer has a public API. If you have a use case for one, "
+                "please open a GitHub issue."
+            ),
+            ScrapyDeprecationWarning,
+            stacklevel=2,
+        )
+        return self._active_size
+
+    @active_size.setter
+    def active_size(self, value: int) -> None:
+        warnings.warn(
+            (
+                "scrapy.core.scraper.Slot.active_size is deprecated, and "
+                "setting it no longer has any effect on request processing."
+            ),
+            ScrapyDeprecationWarning,
+            stacklevel=2,
+        )
+        self._active_size = value
+
+    @property
+    def max_active_size(self) -> int:
+        warnings.warn(
+            (
+                "scrapy.core.scraper.Slot.max_active_size is deprecated. Read "
+                "the RESPONSE_MAX_ACTIVE_SIZE setting instead."
+            ),
+            ScrapyDeprecationWarning,
+            stacklevel=2,
+        )
+        return self._max_active_size
+
+    @max_active_size.setter
+    def max_active_size(self, value: int) -> None:
+        warnings.warn(
+            (
+                "scrapy.core.scraper.Slot.max_active_size is deprecated. Set "
+                "the RESPONSE_MAX_ACTIVE_SIZE setting instead."
+            ),
+            ScrapyDeprecationWarning,
+            stacklevel=2,
+        )
+        self._max_active_size = value
 
     def add_response_request(
         self, result: Response | Failure, request: Request
@@ -79,9 +179,9 @@ class Slot:
         deferred: Deferred[None] = Deferred()
         self.queue.append((result, request, deferred))
         if isinstance(result, Response):
-            self.active_size += max(len(result.body), self.MIN_RESPONSE_SIZE)
+            self._active_size += max(len(result.body), self._MIN_RESPONSE_SIZE)
         else:
-            self.active_size += self.MIN_RESPONSE_SIZE
+            self._active_size += self._MIN_RESPONSE_SIZE
         return deferred
 
     def next_response_request_deferred(self) -> QueueTuple:
@@ -92,15 +192,20 @@ class Slot:
     def finish_response(self, result: Response | Failure, request: Request) -> None:
         self.active.remove(request)
         if isinstance(result, Response):
-            self.active_size -= max(len(result.body), self.MIN_RESPONSE_SIZE)
+            self._active_size -= max(len(result.body), self._MIN_RESPONSE_SIZE)
         else:
-            self.active_size -= self.MIN_RESPONSE_SIZE
+            self._active_size -= self._MIN_RESPONSE_SIZE
 
     def is_idle(self) -> bool:
         return not (self.queue or self.active)
 
     def needs_backout(self) -> bool:
-        return self.active_size > self.max_active_size
+        warnings.warn(
+            "scrapy.core.scraper.Slot.needs_backout is deprecated.",
+            ScrapyDeprecationWarning,
+            stacklevel=2,
+        )
+        return self._active_size > self._max_active_size
 
 
 class Scraper:
@@ -168,7 +273,7 @@ class Scraper:
 
         .. versionadded:: 2.14
         """
-        self.slot = Slot(self.crawler.settings.getint("SCRAPER_SLOT_MAX_ACTIVE_SIZE"))
+        self.slot = Slot()
         if not self.crawler.spider:
             raise RuntimeError(
                 "Scraper.open_spider() called before Crawler.spider is set."

scrapy/settings/default_settings.py

@@ -177,6 +177,8 @@ __all__ = [
     "REFERRER_POLICIES",
     "REFERRER_POLICY",
     "REQUEST_FINGERPRINTER_CLASS",
+    "RESPONSE_MAX_ACTIVE_SIZE",
+    "RESPONSE_ROUGH_SIZE",
     "RETRY_ENABLED",
     "RETRY_EXCEPTIONS",
     "RETRY_GIVE_UP_LOG_LEVEL",
@@ -538,7 +540,10 @@ SCHEDULER_PRIORITY_QUEUE = "scrapy.pqueues.DownloaderAwarePriorityQueue"
 SCHEDULER_START_DISK_QUEUE = "scrapy.squeues.PickleFifoDiskQueue"
 SCHEDULER_START_MEMORY_QUEUE = "scrapy.squeues.FifoMemoryQueue"
 
-SCRAPER_SLOT_MAX_ACTIVE_SIZE = 5000000
+SCRAPER_SLOT_MAX_ACTIVE_SIZE = 5_000_000
+RESPONSE_MAX_ACTIVE_SIZE = 5_000_000
+# None means a share of RESPONSE_MAX_ACTIVE_SIZE based on CONCURRENT_REQUESTS.
+RESPONSE_ROUGH_SIZE = None
 
 SPIDER_CONTRACTS = {}
 SPIDER_CONTRACTS_BASE = {

scrapy/statscollectors.py

@@ -69,7 +69,11 @@ class StatsCollector:
         self._stats = stats
 
     def inc_value(
-        self, key: str, count: int = 1, start: int = 0, spider: Spider | None = None
+        self,
+        key: str,
+        count: float = 1,
+        start: float = 0,
+        spider: Spider | None = None,
     ) -> None:
         d = self._stats
         d[key] = d.setdefault(key, start) + count
@@ -142,7 +146,11 @@ class DummyStatsCollector(StatsCollector):
         pass
 
     def inc_value(
-        self, key: str, count: int = 1, start: int = 0, spider: Spider | None = None
+        self,
+        key: str,
+        count: float = 1,
+        start: float = 0,
+        spider: Spider | None = None,
     ) -> None:
         pass
 

scrapy/utils/engine.py

@@ -24,9 +24,8 @@ def get_engine_status(engine: ExecutionEngine) -> list[tuple[str, Any]]:
         "len(engine._slot.scheduler.mqs)",
         "len(engine.scraper.slot.queue)",
         "len(engine.scraper.slot.active)",
-        "engine.scraper.slot.active_size",
+        "engine.downloader.middleware._total_active_size",
         "engine.scraper.slot.itemproc_size",
-        "engine.scraper.slot.needs_backout()",
     ]
 
     checks: list[tuple[str, Any]] = []

tests/test_core_downloader.py

@@ -15,7 +15,7 @@ from twisted.web.client import Agent, BrowserLikePolicyForHTTPS, readBody
 from twisted.web.client import Response as TxResponse
 
 from scrapy import Request, Spider
-from scrapy.core.downloader import Downloader, Slot, tls
+from scrapy.core.downloader import Downloader, tls
 from scrapy.core.downloader.contextfactory import (
     _load_context_factory_from_settings,
     _ScrapyClientContextFactory,
@@ -43,12 +43,6 @@ if TYPE_CHECKING:
     from scrapy.http import Response
 
 
-class TestSlot:
-    def test_repr(self):
-        slot = Slot(concurrency=8, delay=0.1, randomize_delay=True)
-        assert repr(slot) == "Slot(concurrency=8, delay=0.1, randomize_delay=True)"
-
-
 @pytest.mark.requires_reactor  # this test is related to the Twisted HTTP code
 class TestContextFactoryBase:
     @async_yield_fixture

tests/test_downloader.py

@@ -0,0 +1,598 @@
+import asyncio
+import warnings
+from typing import Any
+
+import pytest
+from twisted.internet.defer import Deferred
+
+from scrapy import Request, Spider
+from scrapy.crawler import Crawler
+from scrapy.exceptions import ScrapyDeprecationWarning
+from scrapy.http import Response
+from scrapy.utils.asyncio import sleep
+from scrapy.utils.defer import maybe_deferred_to_future
+from scrapy.utils.test import get_crawler
+from tests.utils.decorators import coroutine_test
+
+
+def _count_backout_logs(caplog: pytest.LogCaptureFixture) -> int:
+    return sum(
+        1
+        for record in caplog.records
+        if record.levelname == "INFO"
+        and str(record.msg).startswith("Pausing request processing")
+    )
+
+
+def _backout_stats(crawler: Crawler) -> dict[str, Any]:
+    assert crawler.stats
+    return {
+        k: v
+        for k, v in crawler.stats.get_stats().items()
+        if k.startswith("request_backout_seconds/")
+    }
+
+
+class OfflineSpider(Spider):
+    name = "offline"
+    start_urls = ["data:,"]
+
+    def parse(self, response):
+        pass
+
+
+def _assert_scraper_slot_deprecation(
+    warning_messages: pytest.WarningsRecorder, *, ignored: bool = False
+) -> None:
+    """Assert that a crawl emitted exactly one Scrapy deprecation warning, the
+    one about SCRAPER_SLOT_MAX_ACTIVE_SIZE.
+
+    Only Scrapy deprecation warnings are counted: a crawl may emit unrelated
+    warnings (e.g. a ResourceWarning for a socket garbage-collected while the
+    recorder is active), and those must not make the assertion flaky.
+
+    Pass ``ignored=True`` when RESPONSE_MAX_ACTIVE_SIZE is set with an equal or
+    higher priority and therefore SCRAPER_SLOT_MAX_ACTIVE_SIZE is being
+    ignored."""
+    deprecations = [
+        message
+        for message in warning_messages
+        if issubclass(message.category, ScrapyDeprecationWarning)
+    ]
+    assert len(deprecations) == 1
+    if ignored:
+        assert str(deprecations[0].message) == (
+            "The SCRAPER_SLOT_MAX_ACTIVE_SIZE setting is deprecated and is "
+            "being ignored because RESPONSE_MAX_ACTIVE_SIZE is set with an "
+            "equal or higher priority. Remove SCRAPER_SLOT_MAX_ACTIVE_SIZE "
+            "from your settings."
+        )
+    else:
+        assert str(deprecations[0].message) == (
+            "The SCRAPER_SLOT_MAX_ACTIVE_SIZE setting is deprecated, use "
+            "RESPONSE_MAX_ACTIVE_SIZE instead."
+        )
+
+
+class gt:
+    __hash__ = None  # type: ignore[assignment]
+
+    def __init__(self, value: float):
+        self.value = value
+
+    def __eq__(self, other: object) -> bool:
+        return isinstance(other, (int, float)) and other > self.value
+
+    def __repr__(self) -> str:
+        return f">{self.value}"
+
+
+class TestResponseMaxActiveSize:
+    @coroutine_test
+    async def test_default(self):
+        """A crawl without custom settings has its effective response max
+        active size set to 5 000 000, and triggers no deprecation warning."""
+        crawler = get_crawler(OfflineSpider)
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", ScrapyDeprecationWarning)
+            await maybe_deferred_to_future(crawler.crawl())
+        assert crawler.engine
+        assert crawler.engine.downloader.middleware._max_active_size == 5_000_000
+
+    @coroutine_test
+    async def test_custom(self):
+        """Setting RESPONSE_MAX_ACTIVE_SIZE to a custom value changes the
+        effective response max active size."""
+        crawler = get_crawler(
+            OfflineSpider, settings_dict={"RESPONSE_MAX_ACTIVE_SIZE": 0}
+        )
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", ScrapyDeprecationWarning)
+            await maybe_deferred_to_future(crawler.crawl())
+        assert crawler.engine
+        assert crawler.engine.downloader.middleware._max_active_size == 0
+
+    @coroutine_test
+    async def test_deprecated_default(self):
+        """Setting SCRAPER_SLOT_MAX_ACTIVE_SIZE triggers a deprecation warning,
+        even if it is the default value."""
+        crawler = get_crawler(
+            OfflineSpider, settings_dict={"SCRAPER_SLOT_MAX_ACTIVE_SIZE": 5_000_000}
+        )
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            await maybe_deferred_to_future(crawler.crawl())
+        assert crawler.engine
+        assert crawler.engine.downloader.middleware._max_active_size == 5_000_000
+        _assert_scraper_slot_deprecation(warning_messages)
+
+    @coroutine_test
+    async def test_deprecated_custom(self):
+        """Setting SCRAPER_SLOT_MAX_ACTIVE_SIZE to a custom value triggers a
+        deprecation warning, and changes the effective response max active
+        size."""
+        crawler = get_crawler(
+            OfflineSpider, settings_dict={"SCRAPER_SLOT_MAX_ACTIVE_SIZE": 0}
+        )
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            await maybe_deferred_to_future(crawler.crawl())
+        assert crawler.engine
+        assert crawler.engine.downloader.middleware._max_active_size == 0
+        _assert_scraper_slot_deprecation(warning_messages)
+
+    @coroutine_test
+    async def test_both(self):
+        """Setting RESPONSE_MAX_ACTIVE_SIZE and SCRAPER_SLOT_MAX_ACTIVE_SIZE to
+        different values with the same setting priority triggers a deprecation
+        warning about SCRAPER_SLOT_MAX_ACTIVE_SIZE being ignored, and makes
+        the value of RESPONSE_MAX_ACTIVE_SIZE the effective response max active
+        size."""
+        crawler = get_crawler(
+            OfflineSpider,
+            settings_dict={
+                "RESPONSE_MAX_ACTIVE_SIZE": 1,
+                "SCRAPER_SLOT_MAX_ACTIVE_SIZE": 2,
+            },
+        )
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            await maybe_deferred_to_future(crawler.crawl())
+        assert crawler.engine
+        assert crawler.engine.downloader.middleware._max_active_size == 1
+        _assert_scraper_slot_deprecation(warning_messages, ignored=True)
+
+    @coroutine_test
+    async def test_both_deprecated_priority(self):
+        """Setting RESPONSE_MAX_ACTIVE_SIZE and SCRAPER_SLOT_MAX_ACTIVE_SIZE to
+        different values and SCRAPER_SLOT_MAX_ACTIVE_SIZE with a higher
+        priority triggers a deprecation warning about
+        SCRAPER_SLOT_MAX_ACTIVE_SIZE but also makes the value of
+        SCRAPER_SLOT_MAX_ACTIVE_SIZE the effective response max active size."""
+
+        class TestSpider(Spider):
+            name = "test"
+            start_urls = ["data:,"]
+
+            @classmethod
+            def update_settings(cls, settings):
+                settings.set("RESPONSE_MAX_ACTIVE_SIZE", 1, priority=100)
+                settings.set("SCRAPER_SLOT_MAX_ACTIVE_SIZE", 2, priority=101)
+
+            def parse(self, response):
+                pass
+
+        crawler = get_crawler(TestSpider)
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            await maybe_deferred_to_future(crawler.crawl())
+        assert crawler.engine
+        assert crawler.engine.downloader.middleware._max_active_size == 2
+        _assert_scraper_slot_deprecation(warning_messages)
+
+
+class TestResponseRoughSize:
+    @pytest.fixture(autouse=True)
+    def use_caplog(self, caplog):
+        self.caplog = caplog
+
+    @pytest.mark.parametrize(
+        ("settings_dict", "expected"),
+        [
+            # A quarter of RESPONSE_MAX_ACTIVE_SIZE split among
+            # CONCURRENT_REQUESTS requests.
+            ({}, 78125),
+            ({"CONCURRENT_REQUESTS": 100}, 12500),
+            ({"RESPONSE_MAX_ACTIVE_SIZE": 8_000_000}, 125_000),
+            # Unlimited concurrency leaves no number of requests to split it
+            # among.
+            ({"CONCURRENT_REQUESTS": 0}, 0),
+            ({"RESPONSE_MAX_ACTIVE_SIZE": 0}, 0),
+            ({"RESPONSE_ROUGH_SIZE": 1}, 1),
+            ({"RESPONSE_ROUGH_SIZE": 0}, 0),
+        ],
+    )
+    @coroutine_test
+    async def test_value(self, settings_dict, expected):
+        crawler = get_crawler(OfflineSpider, settings_dict=settings_dict)
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", ScrapyDeprecationWarning)
+            await maybe_deferred_to_future(crawler.crawl())
+        assert crawler.engine
+        assert crawler.engine.downloader.middleware._response_rough_size == expected
+
+    @coroutine_test
+    async def test_response_replaces_rough_size(self):
+        """The rough size of a request stops counting as soon as the size of its
+        response is known."""
+        sizes: list[tuple[int, int]] = []
+
+        class DownloaderMiddleware:
+            def __init__(self, crawler):
+                self.crawler = crawler
+
+            @classmethod
+            def from_crawler(cls, crawler):
+                return cls(crawler)
+
+            def process_response(self, request, response):
+                middleware = self.crawler.engine.downloader.middleware
+                sizes.append(
+                    (middleware._rough_active_size, middleware._response_active_size)
+                )
+                return response
+
+        class TestSpider(Spider):
+            name = "test"
+            start_urls = ["data:,a"]
+            custom_settings = {
+                "DOWNLOADER_MIDDLEWARES": {DownloaderMiddleware: 0},
+            }
+
+            def parse(self, response):
+                pass
+
+        crawler = get_crawler(TestSpider)
+        await maybe_deferred_to_future(crawler.crawl())
+
+        assert sizes == [(0, 1)]
+        assert crawler.engine
+        assert crawler.engine.downloader.middleware._rough_sizes == {}
+
+    @pytest.mark.only_asyncio
+    @coroutine_test
+    async def test_same_request_downloaded_twice(self):
+        """The rough size of a request object being downloaded twice at the same
+        time is counted once, so that it is gone once both downloads finish."""
+
+        class TestSpider(Spider):
+            name = "test"
+            start_urls = ["data:,"]
+            custom_settings = {"RESPONSE_ROUGH_SIZE": 1024}
+
+            async def parse(self, response):
+                assert self.crawler.engine
+                request = Request("data:,a")
+                # Only one of the two downloads succeeds, the other one fails
+                # because the request object is already being downloaded.
+                await asyncio.gather(
+                    self.crawler.engine.download_async(request),
+                    self.crawler.engine.download_async(request),
+                    return_exceptions=True,
+                )
+
+        crawler = get_crawler(TestSpider)
+        await maybe_deferred_to_future(crawler.crawl())
+
+        assert crawler.engine
+        middleware = crawler.engine.downloader.middleware
+        assert middleware._rough_sizes == {}
+        assert middleware._rough_active_size == 0
+
+    @coroutine_test
+    async def test_rough_size_triggers_backout(self):
+        """Rough sizes of requests being downloaded count toward the limit, even
+        if their responses turn out to be empty."""
+
+        class SlowDown:
+            """Keeps requests in flight long enough for the engine to check for
+            backout while their rough size is being counted."""
+
+            async def process_request(self, request):
+                await sleep(0.01)
+
+        class TestSpider(Spider):
+            name = "test"
+            start_urls = ["data:,", "data:,"]
+            custom_settings = {
+                "DOWNLOADER_MIDDLEWARES": {SlowDown: 0},
+                "RESPONSE_MAX_ACTIVE_SIZE": 512,
+                "RESPONSE_ROUGH_SIZE": 1024,
+            }
+
+            def parse(self, response):
+                pass
+
+        crawler = get_crawler(TestSpider)
+        self.caplog.clear()
+        with self.caplog.at_level("INFO"):
+            await maybe_deferred_to_future(crawler.crawl())
+
+        assert _count_backout_logs(self.caplog) == 1
+
+        expected_stats = {
+            "request_backout_seconds/response_max_active_size": gt(0),
+            "request_backout_seconds/total": gt(0),
+        }
+        assert _backout_stats(crawler) == expected_stats
+
+
+class TestRequestBackout:
+    @pytest.fixture(autouse=True)
+    def use_caplog(self, caplog):
+        self.caplog = caplog
+
+    @coroutine_test
+    async def test_none(self):
+
+        class TestSpider(Spider):
+            name = "test"
+            start_urls = ["data:,"]
+
+            def parse(self, response):
+                pass
+
+        crawler = get_crawler(TestSpider)
+        self.caplog.clear()
+        with self.caplog.at_level("INFO"):
+            await maybe_deferred_to_future(crawler.crawl())
+
+        assert _count_backout_logs(self.caplog) == 0
+
+        assert _backout_stats(crawler) == {}
+
+    @coroutine_test
+    async def test_concurrency(self):
+
+        class SlowDown:
+            """Downloader middleware that returns a non-instant deferred from
+            process_request, to force need_backout calls to happen at that
+            point.
+
+            The delay is deliberately non-zero so that the concurrency backout
+            state lasts a measurable amount of wall-clock time, which keeps the
+            request_backout_seconds/concurrency stat reliably above 0."""
+
+            def process_request(self, request, spider):
+                from twisted.internet import reactor
+
+                d: Deferred[None] = Deferred()
+                reactor.callLater(0.01, d.callback, None)
+                return d
+
+        class TestSpider(Spider):
+            name = "test"
+            # Several start URLs so that, with CONCURRENT_REQUESTS=1, the engine
+            # reliably attempts to schedule a second request while the first one
+            # is still active, which is what triggers the concurrency backout.
+            start_urls = ["data:,"] * 5
+            custom_settings = {
+                "CONCURRENT_REQUESTS": 1,
+                "DOWNLOADER_MIDDLEWARES": {SlowDown: 0},
+            }
+
+            def parse(self, response):
+                pass
+
+        crawler = get_crawler(TestSpider)
+        self.caplog.clear()
+        with self.caplog.at_level("INFO"):
+            await maybe_deferred_to_future(crawler.crawl())
+
+        assert _count_backout_logs(self.caplog) == 0
+
+        expected_stats = {
+            "request_backout_seconds/concurrency": gt(0),
+            "request_backout_seconds/total": gt(0),
+        }
+        assert _backout_stats(crawler) == expected_stats
+
+    @coroutine_test
+    async def test_response_size(self):
+
+        class TestSpider(Spider):
+            name = "test"
+            start_urls = ["data:,a"]
+            custom_settings = {
+                "RESPONSE_MAX_ACTIVE_SIZE": 1,
+            }
+
+            def parse(self, response):
+                pass
+
+        crawler = get_crawler(TestSpider)
+        self.caplog.clear()
+        with self.caplog.at_level("INFO"):
+            await maybe_deferred_to_future(crawler.crawl())
+
+        assert _count_backout_logs(self.caplog) == 1
+
+        expected_stats = {
+            "request_backout_seconds/response_max_active_size": gt(0),
+            "request_backout_seconds/total": gt(0),
+        }
+        assert _backout_stats(crawler) == expected_stats
+
+    @coroutine_test
+    async def test_response_size_process_request(self):
+
+        class DownloaderMiddleware:
+            def process_request(self, request, spider):
+                return Response("https://example.com", body=b"a")
+
+        class TestSpider(Spider):
+            name = "test"
+            start_urls = ["data:,"]
+            custom_settings = {
+                "DOWNLOADER_MIDDLEWARES": {DownloaderMiddleware: 0},
+                "RESPONSE_MAX_ACTIVE_SIZE": 1,
+            }
+
+            def parse(self, response):
+                pass
+
+        crawler = get_crawler(TestSpider)
+        self.caplog.clear()
+        with self.caplog.at_level("INFO"):
+            await maybe_deferred_to_future(crawler.crawl())
+
+        assert _count_backout_logs(self.caplog) == 1
+
+        expected_stats = {
+            "request_backout_seconds/response_max_active_size": gt(0),
+            "request_backout_seconds/total": gt(0),
+        }
+        assert _backout_stats(crawler) == expected_stats
+
+    @coroutine_test
+    async def test_request_from_process_request(self):
+        """A request returned from process_request does not count toward the
+        limit, even though requests also have a body."""
+
+        class DownloaderMiddleware:
+            def __init__(self):
+                self.replaced = False
+
+            def process_request(self, request):
+                if self.replaced:
+                    return None
+                self.replaced = True
+                return Request("data:,b", body=b"a" * 2000)
+
+        class TestSpider(Spider):
+            name = "test"
+            start_urls = ["data:,"]
+            custom_settings = {
+                "DOWNLOADER_MIDDLEWARES": {DownloaderMiddleware: 0},
+                "RESPONSE_MAX_ACTIVE_SIZE": 1000,
+                "RESPONSE_ROUGH_SIZE": 0,
+            }
+
+            def parse(self, response):
+                pass
+
+        crawler = get_crawler(TestSpider)
+        self.caplog.clear()
+        with self.caplog.at_level("INFO"):
+            await maybe_deferred_to_future(crawler.crawl())
+
+        assert _count_backout_logs(self.caplog) == 0
+
+        assert _backout_stats(crawler) == {}
+
+    @coroutine_test
+    async def test_response_size_process_response(self):
+
+        class DownloaderMiddleware:
+            def process_response(self, request, response, spider):
+                return Response("https://example.com", body=b"a")
+
+        class TestSpider(Spider):
+            name = "test"
+            start_urls = ["data:,"]
+            custom_settings = {
+                "DOWNLOADER_MIDDLEWARES": {DownloaderMiddleware: 0},
+                "RESPONSE_MAX_ACTIVE_SIZE": 1,
+            }
+
+            def parse(self, response):
+                pass
+
+        crawler = get_crawler(TestSpider)
+        self.caplog.clear()
+        with self.caplog.at_level("INFO"):
+            await maybe_deferred_to_future(crawler.crawl())
+
+        assert _count_backout_logs(self.caplog) == 1
+
+        expected_stats = {
+            "request_backout_seconds/response_max_active_size": gt(0),
+            "request_backout_seconds/total": gt(0),
+        }
+        assert _backout_stats(crawler) == expected_stats
+
+    @coroutine_test
+    async def test_response_size_process_exception(self):
+
+        class DownloaderMiddleware1:
+            def process_exception(self, request, exception, spider):
+                return Response("https://example.com", body=b"a")
+
+        class DownloaderMiddleware2:
+            def process_request(self, request, spider):
+                raise ValueError
+
+        class TestSpider(Spider):
+            name = "test"
+            start_urls = ["data:,"]
+            custom_settings = {
+                "DOWNLOADER_MIDDLEWARES": {
+                    DownloaderMiddleware1: 0,
+                    DownloaderMiddleware2: 1,
+                },
+                "RESPONSE_MAX_ACTIVE_SIZE": 1,
+            }
+
+            def parse(self, response):
+                pass
+
+        crawler = get_crawler(TestSpider)
+        self.caplog.clear()
+        with self.caplog.at_level("INFO"):
+            await maybe_deferred_to_future(crawler.crawl())
+
+        assert _count_backout_logs(self.caplog) == 1
+
+        expected_stats = {
+            "request_backout_seconds/response_max_active_size": gt(0),
+            "request_backout_seconds/total": gt(0),
+        }
+        assert _backout_stats(crawler) == expected_stats
+
+    @coroutine_test
+    async def test_response_size_download(self):
+        """Ensure that responses from engine.download calls are also taken into
+        account for the RESPONSE_MAX_ACTIVE_SIZE setting."""
+
+        class SlowDown:
+            """Item pipeline that returns a non-instant deferred, to force
+            need_backout calls to happen at that point."""
+
+            def process_item(self, item, spider):
+                from twisted.internet import reactor
+
+                d: Deferred[dict[Any, Any]] = Deferred()
+                reactor.callLater(0, d.callback, {})
+                return d
+
+        class TestSpider(Spider):
+            name = "test"
+            start_urls = ["data:,"]
+            custom_settings = {
+                "ITEM_PIPELINES": {SlowDown: 0},
+                "RESPONSE_MAX_ACTIVE_SIZE": 1,
+            }
+
+            async def parse(self, response):
+                assert self.crawler.engine
+                response = await self.crawler.engine.download(Request("data:,a"))
+                yield {"response": response}
+
+        crawler = get_crawler(TestSpider)
+        self.caplog.clear()
+        with self.caplog.at_level("INFO"):
+            await maybe_deferred_to_future(crawler.crawl())
+
+        assert _count_backout_logs(self.caplog) == 1
+
+        expected_stats = {
+            "request_backout_seconds/response_max_active_size": gt(0),
+            "request_backout_seconds/total": gt(0),
+        }
+        assert _backout_stats(crawler) == expected_stats

tests/test_engine.py

@@ -15,7 +15,7 @@ from scrapy.core.scheduler import BaseScheduler
 from scrapy.exceptions import CloseSpider, IgnoreRequest
 from scrapy.http import Request
 from scrapy.spiders import Spider
-from scrapy.utils.defer import _schedule_coro, deferred_from_coro
+from scrapy.utils.defer import deferred_from_coro
 from scrapy.utils.misc import build_from_crawler
 from scrapy.utils.spider import DefaultSpider
 from scrapy.utils.test import get_crawler
@@ -131,10 +131,11 @@ class TestEngine(TestEngineBase):
         e = ExecutionEngine(crawler, lambda _: None)
         crawler.engine = e
         yield deferred_from_coro(e.open_spider_async())
-        _schedule_coro(e.start_async())
+        start_deferred = deferred_from_coro(e.start_async())
         with pytest.raises(RuntimeError, match="Engine already running"):
             yield deferred_from_coro(e.start_async())
         yield deferred_from_coro(e.stop_async())
+        yield start_deferred
 
     @pytest.mark.only_asyncio
     @coroutine_test

tests/test_scraper.py

@@ -0,0 +1,190 @@
+import warnings
+
+import pytest
+
+from scrapy import Spider
+from scrapy.core.scraper import Slot
+from scrapy.exceptions import ScrapyDeprecationWarning
+from scrapy.utils.defer import maybe_deferred_to_future
+from scrapy.utils.test import get_crawler
+from tests.utils.decorators import coroutine_test
+
+
+class TestScraper:
+    @coroutine_test
+    async def test_crawl(self):
+        """A crawl should not trigger any deprecation warning."""
+        outcome = {}
+
+        class TestSpider(Spider):
+            name = "test"
+            start_urls = ["data:,"]
+
+            def parse(self, response):
+                assert self.crawler.engine
+                slot = self.crawler.engine.scraper.slot
+                assert slot
+                with pytest.warns(ScrapyDeprecationWarning):
+                    outcome["active_size"] = slot.active_size
+
+        crawler = get_crawler(
+            TestSpider, settings_dict={"TELNETCONSOLE_ENABLED": False}
+        )
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", ScrapyDeprecationWarning)
+            await maybe_deferred_to_future(crawler.crawl())
+
+        assert crawler.engine
+        slot = crawler.engine.scraper.slot
+        assert slot
+        with pytest.warns(ScrapyDeprecationWarning):
+            expected = slot.MIN_RESPONSE_SIZE
+        assert outcome["active_size"] == expected
+
+
+class TestSlot:
+    def setup_method(self):
+        self._saved_min_response_size = Slot._MIN_RESPONSE_SIZE
+
+    def teardown_method(self):
+        Slot._MIN_RESPONSE_SIZE = self._saved_min_response_size
+
+    def test_min_response_time_read(self):
+        slot = Slot()
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            actual = slot.MIN_RESPONSE_SIZE
+        assert actual == 1024
+        assert len(warning_messages) == 1
+        assert (
+            str(warning_messages[0].message)
+            == "scrapy.core.scraper.Slot.MIN_RESPONSE_SIZE is deprecated."
+        )
+
+    def test_min_response_time_write(self):
+        slot = Slot()
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            slot.MIN_RESPONSE_SIZE = 0
+        with pytest.warns(ScrapyDeprecationWarning):
+            assert slot.MIN_RESPONSE_SIZE == 0
+        assert len(warning_messages) == 1
+        assert (
+            str(warning_messages[0].message)
+            == "scrapy.core.scraper.Slot.MIN_RESPONSE_SIZE is deprecated."
+        )
+
+    def test_min_response_size_class_read(self):
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            actual = Slot.MIN_RESPONSE_SIZE
+        assert actual == 1024
+        assert len(warning_messages) == 1
+        assert (
+            str(warning_messages[0].message)
+            == "scrapy.core.scraper.Slot.MIN_RESPONSE_SIZE is deprecated."
+        )
+
+    def test_min_response_size_class_write(self):
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            Slot.MIN_RESPONSE_SIZE = 0
+        with pytest.warns(ScrapyDeprecationWarning):
+            assert Slot.MIN_RESPONSE_SIZE == 0
+        assert len(warning_messages) == 1
+        assert (
+            str(warning_messages[0].message)
+            == "scrapy.core.scraper.Slot.MIN_RESPONSE_SIZE is deprecated."
+        )
+
+    def test_slot_init_max_active_size_default(self):
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            slot = Slot(max_active_size=5_000_000)
+        with pytest.warns(ScrapyDeprecationWarning):
+            assert slot.max_active_size == 5_000_000
+        assert len(warning_messages) == 1
+        assert str(warning_messages[0].message) == (
+            "The max_active_size parameter of scrapy.core.scraper.Slot is "
+            "deprecated. Use the RESPONSE_MAX_ACTIVE_SIZE setting instead."
+        )
+
+    def test_slot_init_max_active_size_custom(self):
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            slot = Slot(max_active_size=0)
+        with pytest.warns(ScrapyDeprecationWarning):
+            assert slot.max_active_size == 0
+        assert len(warning_messages) == 1
+        assert str(warning_messages[0].message) == (
+            "The max_active_size parameter of scrapy.core.scraper.Slot is "
+            "deprecated. Use the RESPONSE_MAX_ACTIVE_SIZE setting instead."
+        )
+
+    def test_max_active_size_read(self):
+        slot = Slot()
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            actual = slot.max_active_size
+        assert actual == 5_000_000
+        assert len(warning_messages) == 1
+        assert str(warning_messages[0].message) == (
+            "scrapy.core.scraper.Slot.max_active_size is deprecated. Read "
+            "the RESPONSE_MAX_ACTIVE_SIZE setting instead."
+        )
+
+    def test_max_active_size_write(self):
+        slot = Slot()
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            slot.max_active_size = 0
+        with pytest.warns(ScrapyDeprecationWarning):
+            assert slot.max_active_size == 0
+        assert len(warning_messages) == 1
+        assert str(warning_messages[0].message) == (
+            "scrapy.core.scraper.Slot.max_active_size is deprecated. Set "
+            "the RESPONSE_MAX_ACTIVE_SIZE setting instead."
+        )
+
+    def test_active_size_read(self):
+        slot = Slot()
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            actual = slot.active_size
+        assert actual == 0
+        assert len(warning_messages) == 1
+        assert str(warning_messages[0].message) == (
+            "scrapy.core.scraper.Slot.active_size is deprecated. The size of "
+            "responses in memory is now tracked by the downloader, and no "
+            "longer has a public API. If you have a use case for one, please "
+            "open a GitHub issue."
+        )
+
+    def test_active_size_write(self):
+        slot = Slot()
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            slot.active_size = 1
+        with pytest.warns(ScrapyDeprecationWarning):
+            assert slot.active_size == 1
+        assert len(warning_messages) == 1
+        assert str(warning_messages[0].message) == (
+            "scrapy.core.scraper.Slot.active_size is deprecated, and setting "
+            "it no longer has any effect on request processing."
+        )
+
+    def test_needs_backout_false(self):
+        slot = Slot()
+        with pytest.warns(ScrapyDeprecationWarning):
+            slot.active_size = 5_000_000
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            actual = slot.needs_backout()
+        assert actual is False
+        assert len(warning_messages) == 1
+        assert (
+            str(warning_messages[0].message)
+            == "scrapy.core.scraper.Slot.needs_backout is deprecated."
+        )
+
+    def test_needs_backout_true(self):
+        slot = Slot()
+        with pytest.warns(ScrapyDeprecationWarning):
+            slot.active_size = 5_000_001
+        with pytest.warns(ScrapyDeprecationWarning) as warning_messages:
+            actual = slot.needs_backout()
+        assert actual is True
+        assert len(warning_messages) == 1
+        assert (
+            str(warning_messages[0].message)
+            == "scrapy.core.scraper.Slot.needs_backout is deprecated."
+        )