Merge 38fe2751b3 into d8ba1571e7

docs/news.rst

@@ -5805,9 +5805,9 @@ New features
 *   A new ``DNS_RESOLVER`` setting allows enabling IPv6 support
     (:issue:`1031`, :issue:`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 (:issue:`1410`, :issue:`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 (:issue:`1410`, :issue:`3551`)
 
 *   A new :setting:`TWISTED_REACTOR` setting allows customizing the
     :mod:`~twisted.internet.reactor` that Scrapy uses, allowing to

docs/topics/practices.rst

@@ -409,6 +409,35 @@ run:
 Because :setting:`SPIDER_MODULES` is a list setting, you can include multiple
 modules by separating them with commas.
 
+.. _crawl-optimization:
+
+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/request-response.rst

@@ -734,6 +734,7 @@ Those are:
 * :reqmeta:`redirect_reasons`
 * :reqmeta:`redirect_urls`
 * :reqmeta:`referrer_policy`
+* :reqmeta:`response_rough_size`
 * :reqmeta:`verbatim_url`
 
 .. reqmeta:: bindaddress
@@ -848,6 +849,20 @@ The meta key is used set retry times per request. When set, the
 :reqmeta:`max_retry_times` meta key takes higher precedence over the
 :setting:`RETRY_TIMES` setting.
 
+.. reqmeta:: response_rough_size
+
+response_rough_size
+-------------------
+
+Overrides the :setting:`RESPONSE_ROUGH_SIZE` setting for this specific request.
+
+The value is the estimated size (in bytes) to count toward
+:setting:`RESPONSE_MAX_ACTIVE_SIZE` while this request is being downloaded,
+before its actual response size is known. Set to ``0`` to disable rough-size
+counting for this request.
+
+.. versionadded:: VERSION
+
 .. reqmeta:: verbatim_url
 
 verbatim_url

docs/topics/settings.rst

@@ -1787,6 +1787,71 @@ 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.
+
+When the total exceeds this value, Scrapy pauses sending new requests to the
+downloader until it drops below the limit.
+
+If you set this to ``0``, the limit is disabled.
+
+Setting this to a lower value reduces memory usage at the cost of crawl speed.
+Setting this to a higher value (or disabling it) improves crawl speed but may
+cause memory issues when responses are large.
+
+When the limit is first reached, Scrapy logs an info-level message explaining
+the situation. Check the ``request_backout_seconds/response_max_active_size``
+stat to see how long request processing has been paused due to this limit over
+the course of a crawl.
+
+.. caution::
+
+    If your code stores strong references to :class:`~scrapy.http.Response`
+    objects (e.g. in a scheduled request's meta or in a component attribute),
+    the garbage collector cannot free them, and the total active size may not
+    drop below the limit. In that case your crawl might get stuck indefinitely.
+    Either avoid storing such references, or set this to ``0`` to disable the
+    limit.
+
+    To check whether your crawl is stuck due to this, connect to the
+    :ref:`telnet console <topics-telnetconsole>` and run ``prefs()`` to see
+    the count of live :class:`~scrapy.http.Response` objects. If that count
+    is large and not decreasing, you likely have strong response references.
+    See :ref:`topics-leaks` for details.
+
+.. versionadded:: VERSION
+
+.. setting:: RESPONSE_ROUGH_SIZE
+
+RESPONSE_ROUGH_SIZE
+-------------------
+
+Default: ``131072`` (128 kiB)
+
+Estimated size (in bytes) to count toward :setting:`RESPONSE_MAX_ACTIVE_SIZE`
+for each request that is currently being downloaded, before its actual response
+size is known.
+
+This allows :setting:`RESPONSE_MAX_ACTIVE_SIZE` to provide backpressure based
+on the number of concurrent in-flight requests, not just already-received
+responses. Once the response arrives, its actual body size is counted instead.
+
+You can override this value on a per-request basis via the
+:reqmeta:`response_rough_size` request meta key.
+
+.. versionadded:: VERSION
+
 .. setting:: ROBOTSTXT_OBEY
 
 ROBOTSTXT_OBEY
@@ -1937,19 +2002,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

scrapy/_classutilities.py

@@ -0,0 +1,102 @@
+# https://github.com/david-salac/classutilities/issues/1
+# Modified copy of
+# https://github.com/david-salac/classutilities/blob/a6e4a86331936d432afaa454ed4c963528165a61/src/classutilities/classproperty.py
+
+# Allows creating a class level property
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+
+class ClassPropertyContainer:
+    """
+    Allows creating a class level property (functionality for
+    decorator).
+    """
+
+    def __init__(self, prop_get: Any, prop_set: Any = None):
+        """
+        Container that allows having a class property decorator.
+        :param prop_get: Class property getter.
+        :param prop_set: Class property setter.
+        """
+        self.prop_get: Any = prop_get
+        self.prop_set: Any = prop_set
+
+    def __get__(self, obj: Any, cls: type | None = None) -> Any:
+        """
+        Return the value of the class property.
+        :param obj: Instance of the class.
+        :param cls: Type of the class.
+        :return: Value of the class property.
+        """
+        if cls is None:  # pragma: no cover
+            cls = type(obj)
+        return self.prop_get.__get__(obj, cls)()
+
+    def __set__(self, obj: Any, value: Any) -> None:
+        """
+        Set the value of the class property.
+        :param obj: Instance of the class.
+        :param value: A value to be set.
+        """
+        if not self.prop_set:  # pragma: no cover
+            raise AttributeError("cannot set attribute")
+        _type: type = type(obj)
+        if _type == ClassPropertyMetaClass:
+            _type = obj
+        self.prop_set.__get__(obj, _type)(value)
+
+    def setter(
+        self,
+        func: Callable[..., Any] | classmethod[Any, Any, Any] | staticmethod[Any, Any],
+    ) -> ClassPropertyContainer:
+        """
+        Allows creating setter in a property like way.
+        :param func: Setter function.
+        :return: Setter object for the decorator.
+        """
+        if not isinstance(func, (classmethod, staticmethod)):
+            func = classmethod(func)
+        self.prop_set = func
+        return self
+
+
+def classproperty(
+    func: Callable[..., Any] | classmethod[Any, Any, Any] | staticmethod[Any, Any],
+) -> ClassPropertyContainer:
+    """
+    Create a decorator for a class level property.
+    :param func: This class method is decorated.
+    :return: Modified class method behaving like a class property.
+    """
+    if not isinstance(func, (classmethod, staticmethod)):
+        # The method must be a classmethod (or staticmethod)
+        func = classmethod(func)
+    return ClassPropertyContainer(func)
+
+
+class ClassPropertyMetaClass(type):
+    """
+    Metaclass that allows creating a standard setter.
+    """
+
+    def __setattr__(cls, key: str, value: Any) -> None:
+        """Overloads setter for class"""
+        obj = None
+        if key in cls.__dict__:
+            obj = cls.__dict__.get(key)
+        if obj and isinstance(obj, ClassPropertyContainer):
+            return obj.__set__(cls, value)
+
+        return super().__setattr__(key, value)
+
+
+class ClassPropertiesMixin(metaclass=ClassPropertyMetaClass):
+    """
+    This mixin allows using class properties setter (getter works
+    correctly even without this mixin)
+    """

scrapy/core/downloader/__init__.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
+import gc
 import random
 from collections import deque
 from dataclasses import dataclass, field
 from datetime import datetime
-from time import monotonic
+from logging import getLogger
+from time import monotonic, time
 from typing import TYPE_CHECKING, Any
+from warnings import warn
 
 from twisted.internet.defer import Deferred, inlineCallbacks
 from twisted.python.failure import Failure
@@ -13,6 +16,7 @@ from twisted.python.failure import Failure
 from scrapy import Request, Spider, signals
 from scrapy.core.downloader.handlers import DownloadHandlers
 from scrapy.core.downloader.middleware import DownloaderMiddlewareManager
+from scrapy.exceptions import ScrapyDeprecationWarning
 from scrapy.resolver import dnscache
 from scrapy.utils.asyncio import (
     AsyncioLoopingCall,
@@ -40,8 +44,10 @@ if TYPE_CHECKING:
     from scrapy.settings import BaseSettings
     from scrapy.signalmanager import SignalManager
 
+logger = getLogger(__name__)
 
-@dataclass(slots=True, eq=False)
+
+@dataclass(slots=True, eq=False, repr=False)
 class Slot:
     """Downloader slot"""
 
@@ -49,13 +55,20 @@ class Slot:
     delay: float
     randomize_delay: bool
 
-    active: set[Request] = field(default_factory=set, init=False, repr=False)
+    active: set[Request] = field(default_factory=set, init=False)
     queue: deque[tuple[Request, Deferred[Response]]] = field(
-        default_factory=deque, init=False, repr=False
+        default_factory=deque, init=False
     )
-    transferring: set[Request] = field(default_factory=set, init=False, repr=False)
-    lastseen: float = field(default=0, init=False, repr=False)
-    latercall: CallLaterResult | None = field(default=None, init=False, repr=False)
+    transferring: set[Request] = field(default_factory=set, init=False)
+    lastseen: float = field(default=0, init=False)
+    latercall: CallLaterResult | None = field(default=None, init=False)
+
+    def __repr__(self) -> str:
+        return (
+            f"Slot(concurrency={self.concurrency!r}, "
+            f"delay={self.delay:.2f}, "
+            f"randomize_delay={self.randomize_delay!r})"
+        )
 
     def free_transfer_slots(self) -> int:
         return self.concurrency - len(self.transferring)
@@ -120,6 +133,49 @@ 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)
+
+        deprecated_setting_priority = self.settings.getpriority(
+            "SCRAPER_SLOT_MAX_ACTIVE_SIZE"
+        )
+        assert deprecated_setting_priority is not None
+        setting_priority = self.settings.getpriority("RESPONSE_MAX_ACTIVE_SIZE")
+        assert setting_priority is not None
+        if deprecated_setting_priority > 0:
+            if setting_priority >= deprecated_setting_priority:
+                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,
+                )
+                self._response_max_active_size = self.settings.getint(
+                    "RESPONSE_MAX_ACTIVE_SIZE"
+                )
+            else:
+                warn(
+                    (
+                        "The SCRAPER_SLOT_MAX_ACTIVE_SIZE setting is deprecated, "
+                        "use RESPONSE_MAX_ACTIVE_SIZE instead."
+                    ),
+                    ScrapyDeprecationWarning,
+                    stacklevel=2,
+                )
+                self._response_max_active_size = self.settings.getint(
+                    "SCRAPER_SLOT_MAX_ACTIVE_SIZE"
+                )
+        else:
+            self._response_max_active_size = self.settings.getint(
+                "RESPONSE_MAX_ACTIVE_SIZE"
+            )
+        self._response_max_active_size_warned = False
 
     @inlineCallbacks
     @_warn_spider_arg
@@ -127,6 +183,7 @@ class Downloader:
         self, request: Request, spider: Spider | None = None
     ) -> Generator[Deferred[Any], Any, Response | Request]:
         self.active.add(request)
+        rough_size = self.middleware._count_rough_size(request)
         try:
             result: Response | Request = yield (
                 deferred_from_coro(
@@ -136,9 +193,61 @@ class Downloader:
             return result
         finally:
             self.active.remove(request)
+            self.middleware._discount_rough_size(rough_size)
+
+    def _record_backout(self, reason: str | None) -> None:
+        last_reason, last_reason_start_time = self._last_backout
+        if last_reason == reason:
+            return
+        current_time = time()
+        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:
-        return len(self.active) >= self.total_concurrency
+        if len(self.active) >= self.total_concurrency:
+            self._record_backout("concurrency")
+            return True
+        if (
+            self._response_max_active_size
+            and self.middleware.total_active_size >= self._response_max_active_size
+        ):
+            if not self._response_max_active_size_warned:
+                self._response_max_active_size_warned = True
+                logger.info(
+                    f"The active response size, i.e. the total size of all "
+                    f"bodies from responses that have been processed by "
+                    f"downloader middlewares and remain in memory, plus the "
+                    f"rough sizes of in-flight requests, is "
+                    f"{self.middleware.total_active_size} B. The "
+                    f"RESPONSE_MAX_ACTIVE_SIZE setting sets its maximum value "
+                    f"at {self._response_max_active_size} B. No more requests "
+                    f"will be processed until active response size lowers. If "
+                    f"your memory allows it, you may increase "
+                    f"RESPONSE_MAX_ACTIVE_SIZE, which should increase your "
+                    f"crawl speed. If your code keeps non-weak references to "
+                    f"Response objects, e.g. in (scheduled) requests or in a "
+                    f"container within a component, your crawl might get "
+                    f"stuck indefinitely; you can set "
+                    f"RESPONSE_MAX_ACTIVE_SIZE to 0 to disable this limit, "
+                    f"but then your code might run out of memory. This "
+                    f"message will only appear the first time this happens. "
+                    f"To learn how often request processing has been paused "
+                    f"during a crawl for this reason, see the "
+                    f"request_backout_seconds/response_max_active_size stat."
+                )
+            self._record_backout("response_max_active_size")
+            # Force the garbage collection of response objects. Necessary for
+            # PyPy, which is lazier when it comes to garbage collection.
+            gc.collect()
+            return True
+        self._record_backout(None)
+        return False
 
     @_warn_spider_arg
     def _get_slot(
@@ -265,6 +374,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
@@ -34,6 +35,16 @@ if TYPE_CHECKING:
 class DownloaderMiddlewareManager(MiddlewareManager):
     component_name = "downloader middleware"
 
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.response_active_size = 0
+        self._tracked_responses: WeakSet[Response] = WeakSet()
+        self._rough_active_size = 0
+        assert self.crawler is not None
+        self._response_rough_size: int = self.crawler.settings.getint(
+            "RESPONSE_ROUGH_SIZE"
+        )
+
     @classmethod
     def _get_mwlist_from_settings(cls, settings: BaseSettings) -> list[Any]:
         return build_component_list(
@@ -51,6 +62,30 @@ 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:
+        """Sum of sizes of tracked responses and rough sizes of in-flight requests."""
+        return self.response_active_size + self._rough_active_size
+
+    def _count_rough_size(self, request: Request) -> int:
+        size: int = request.meta.get("response_rough_size", self._response_rough_size)
+        self._rough_active_size += size
+        return size
+
+    def _discount_rough_size(self, size: int) -> None:
+        self._rough_active_size -= size
+
+    def _count_response_size(self, response: Response) -> None:
+        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 +145,13 @@ class DownloaderMiddlewareManager(MiddlewareManager):
                     f"Request, got {response.__class__.__name__}"
                 )
             if response:
+                if isinstance(response, Response):
+                    self._count_response_size(response)
                 return response
-        return await download_func(request)
+        result = await download_func(request)
+        if isinstance(result, Response):
+            self._count_response_size(result)
+        return result
 
     async def _process_response(
         self, response: Response | Request, request: Request
@@ -135,6 +175,7 @@ class DownloaderMiddlewareManager(MiddlewareManager):
                 )
             if isinstance(response, Request):
                 return response
+            self._count_response_size(response)
         return response
 
     async def _process_exception(
@@ -152,5 +193,7 @@ class DownloaderMiddlewareManager(MiddlewareManager):
                     f"Request, got {type(response)}"
                 )
             if response:
+                if isinstance(response, Response):
+                    self._count_response_size(response)
                 return response
         raise exception

scrapy/core/engine.py

@@ -350,7 +350,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

@@ -13,6 +13,7 @@ from twisted.internet.defer import Deferred, inlineCallbacks
 from twisted.python.failure import Failure
 
 from scrapy import Spider, signals
+from scrapy._classutilities import ClassPropertiesMixin, classproperty
 from scrapy.core.spidermw import SpiderMiddlewareManager
 from scrapy.exceptions import (
     CloseSpider,
@@ -55,18 +56,106 @@ _T = TypeVar("_T")
 QueueTuple: TypeAlias = tuple[Response | Failure, Request, Deferred[None]]
 
 
-class Slot:
+_UNSET = object()
+
+
+class Slot(ClassPropertiesMixin):
     """Scraper slot (one per running spider)"""
 
-    MIN_RESPONSE_SIZE = 1024
+    _MIN_RESPONSE_SIZE = 1024
 
-    def __init__(self, max_active_size: int = 5000000):
-        self.max_active_size: int = max_active_size
+    @classproperty
+    def MIN_RESPONSE_SIZE(cls):  # pylint: disable=no-self-argument
+        warnings.warn(
+            "scrapy.core.scraper.Slot.MIN_RESPONSE_SIZE is deprecated.",
+            ScrapyDeprecationWarning,
+            stacklevel=2,
+        )
+        return cls._MIN_RESPONSE_SIZE
+
+    @MIN_RESPONSE_SIZE.setter  # type: ignore[no-redef]
+    def MIN_RESPONSE_SIZE(cls, value):  # pylint: disable=no-self-argument
+        warnings.warn(
+            "scrapy.core.scraper.Slot.MIN_RESPONSE_SIZE is deprecated.",
+            ScrapyDeprecationWarning,
+            stacklevel=2,
+        )
+        cls._MIN_RESPONSE_SIZE = value
+
+    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. Read "
+                "scrapy.core.downloader.DownloaderMiddlewareManager.response_active_size "
+                "instead."
+            ),
+            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. "
+                "scrapy.core.downloader.DownloaderMiddlewareManager.response_active_size "
+                "might work as a replacement, but modifying that attribute "
+                "might not be a good idea. If you have a use case for it, you "
+                "might want to bring it up in a GitHub issue, to discuss with "
+                "Scrapy developers if there is a better approach, or some "
+                "change we could implement in Scrapy to improve support for "
+                "your use case."
+            ),
+            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
@@ -75,9 +164,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:
@@ -88,15 +177,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:
@@ -165,7 +259,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

@@ -176,6 +176,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",
@@ -534,7 +536,9 @@ 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
+RESPONSE_ROUGH_SIZE = 131072
 
 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
@@ -123,7 +127,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/defer.py

@@ -537,9 +537,14 @@ def _schedule_coro(coro: Coroutine[Any, Any, Any]) -> None:
     alternative is calling :func:`scrapy.utils.defer.deferred_from_coro`,
     keeping the result, and adding proper exception handling (e.g. errbacks) to
     it.
+
+    In both asyncio and non-asyncio modes the coroutine starts in the next
+    event-loop iteration, never in the current call stack.
     """
     if not is_asyncio_available():
-        Deferred.fromCoroutine(coro)
+        from twisted.internet import reactor
+
+        reactor.callLater(0, Deferred.fromCoroutine, coro)
         return
     loop = asyncio.get_event_loop()
     loop.create_task(coro)  # noqa: RUF006

tests/test_core_downloader.py

@@ -14,7 +14,7 @@ from twisted.web import server, static
 from twisted.web.client import Agent, BrowserLikePolicyForHTTPS, readBody
 from twisted.web.client import Response as TxResponse
 
-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,
@@ -39,12 +39,6 @@ if TYPE_CHECKING:
     from twisted.web.iweb import IBodyProducer
 
 
-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,586 @@
+import warnings
+
+import pytest
+from twisted.internet.defer import Deferred
+
+from scrapy import Request, Spider
+from scrapy.core.downloader import Slot
+from scrapy.exceptions import ScrapyDeprecationWarning
+from scrapy.http import Response
+from scrapy.utils.defer import maybe_deferred_to_future
+from scrapy.utils.test import get_crawler
+from tests.utils.decorators import coroutine_test
+
+
+class TestSlot:
+    def test_repr(self):
+        slot = Slot(concurrency=8, delay=0.1, randomize_delay=True)
+        assert repr(slot) == "Slot(concurrency=8, delay=0.10, randomize_delay=True)"
+
+
+class OfflineSpider(Spider):
+    name = "offline"
+    start_urls = ["data:,"]
+
+    def parse(self, response):
+        pass
+
+
+def _assert_scraper_slot_deprecation(warning_messages, *, ignored=False):
+    """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):
+        self.value = value
+
+    def __eq__(self, other):
+        return other > self.value
+
+    def __repr__(self):
+        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.downloader._response_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.downloader._response_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.downloader._response_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.downloader._response_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.downloader._response_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.downloader._response_max_active_size == 2
+        _assert_scraper_slot_deprecation(warning_messages)
+
+
+class TestResponseRoughSize:
+    @pytest.fixture(autouse=True)
+    def use_caplog(self, caplog):
+        self.caplog = caplog
+
+    @coroutine_test
+    async def test_default(self):
+        crawler = get_crawler(OfflineSpider)
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", ScrapyDeprecationWarning)
+            await maybe_deferred_to_future(crawler.crawl())
+        assert crawler.engine.downloader.middleware._response_rough_size == 131072
+
+    @coroutine_test
+    async def test_custom(self):
+        """Setting RESPONSE_ROUGH_SIZE to a custom value changes the rough size."""
+        crawler = get_crawler(OfflineSpider, settings_dict={"RESPONSE_ROUGH_SIZE": 0})
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", ScrapyDeprecationWarning)
+            await maybe_deferred_to_future(crawler.crawl())
+        assert crawler.engine.downloader.middleware._response_rough_size == 0
+
+    @coroutine_test
+    async def test_rough_size_per_request(self):
+        """response_rough_size meta key overrides RESPONSE_ROUGH_SIZE per request.
+
+        A low RESPONSE_MAX_ACTIVE_SIZE is set so that only requests with the
+        custom rough size of 1 pass; without the override the default 131072
+        would trigger backout and only one request would be downloaded."""
+
+        class TestSpider(Spider):
+            name = "test"
+            custom_settings = {
+                "RESPONSE_MAX_ACTIVE_SIZE": 512,
+            }
+
+            async def start(self):
+                yield Request("data:,a", meta={"response_rough_size": 1})
+                yield Request("data:,b")
+
+            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())
+
+        active_size_log_count = sum(
+            1
+            for r in self.caplog.records
+            if str(r.msg).startswith("The active response size")
+            and r.levelname == "INFO"
+        )
+        assert active_size_log_count == 1
+
+    @coroutine_test
+    async def test_rough_size_triggers_backout(self):
+        """Rough sizes of in-flight requests count toward the backpressure limit.
+
+        With RESPONSE_MAX_ACTIVE_SIZE=512 and RESPONSE_ROUGH_SIZE=1024, even a
+        response with an empty body should trigger the backout log."""
+
+        class TestSpider(Spider):
+            name = "test"
+            start_urls = ["data:,", "data:,"]
+            custom_settings = {
+                "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())
+
+        matching_log_count = 0
+        for log_record in self.caplog.records:
+            if (
+                str(log_record.msg).startswith("The active response size")
+                and log_record.levelname == "INFO"
+            ):
+                matching_log_count += 1
+        assert matching_log_count == 1
+
+        expected_stats = {
+            "request_backout_seconds/response_max_active_size": gt(0),
+            "request_backout_seconds/total": gt(0),
+        }
+        actual_stats = {
+            k: v
+            for k, v in crawler.stats.get_stats().items()
+            if k.startswith("request_backout_seconds/")
+        }
+        assert expected_stats == actual_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())
+
+        matching_log_count = 0
+        for log_record in self.caplog.records:
+            if (
+                str(log_record.msg).startswith("The active response size")
+                and log_record.levelname == "INFO"
+            ):
+                matching_log_count += 1
+        assert matching_log_count == 0
+
+        stats = {
+            k: v
+            for k, v in crawler.stats.get_stats().items()
+            if k.startswith("request_backout_seconds/")
+        }
+        assert stats == {}
+
+    @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()
+                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())
+
+        matching_log_count = 0
+        for log_record in self.caplog.records:
+            if (
+                str(log_record.msg).startswith("The active response size")
+                and log_record.levelname == "INFO"
+            ):
+                matching_log_count += 1
+        assert matching_log_count == 0
+
+        expected_stats = {
+            "request_backout_seconds/concurrency": gt(0),
+            "request_backout_seconds/total": gt(0),
+        }
+        actual_stats = {
+            k: v
+            for k, v in crawler.stats.get_stats().items()
+            if k.startswith("request_backout_seconds/")
+        }
+        assert expected_stats == actual_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())
+
+        matching_log_count = 0
+        for log_record in self.caplog.records:
+            if (
+                str(log_record.msg).startswith("The active response size")
+                and log_record.levelname == "INFO"
+            ):
+                matching_log_count += 1
+        assert matching_log_count == 1
+
+        expected_stats = {
+            "request_backout_seconds/response_max_active_size": gt(0),
+            "request_backout_seconds/total": gt(0),
+        }
+        actual_stats = {
+            k: v
+            for k, v in crawler.stats.get_stats().items()
+            if k.startswith("request_backout_seconds/")
+        }
+        assert expected_stats == actual_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())
+
+        matching_log_count = 0
+        for log_record in self.caplog.records:
+            if (
+                str(log_record.msg).startswith("The active response size")
+                and log_record.levelname == "INFO"
+            ):
+                matching_log_count += 1
+        assert matching_log_count == 1
+
+        expected_stats = {
+            "request_backout_seconds/response_max_active_size": gt(0),
+            "request_backout_seconds/total": gt(0),
+        }
+        actual_stats = {
+            k: v
+            for k, v in crawler.stats.get_stats().items()
+            if k.startswith("request_backout_seconds/")
+        }
+        assert expected_stats == actual_stats
+
+    @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())
+
+        matching_log_count = 0
+        for log_record in self.caplog.records:
+            if (
+                str(log_record.msg).startswith("The active response size")
+                and log_record.levelname == "INFO"
+            ):
+                matching_log_count += 1
+        assert matching_log_count == 1
+
+        expected_stats = {
+            "request_backout_seconds/response_max_active_size": gt(0),
+            "request_backout_seconds/total": gt(0),
+        }
+        actual_stats = {
+            k: v
+            for k, v in crawler.stats.get_stats().items()
+            if k.startswith("request_backout_seconds/")
+        }
+        assert expected_stats == actual_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())
+
+        matching_log_count = 0
+        for log_record in self.caplog.records:
+            if (
+                str(log_record.msg).startswith("The active response size")
+                and log_record.levelname == "INFO"
+            ):
+                matching_log_count += 1
+        assert matching_log_count == 1
+
+        expected_stats = {
+            "request_backout_seconds/response_max_active_size": gt(0),
+            "request_backout_seconds/total": gt(0),
+        }
+        actual_stats = {
+            k: v
+            for k, v in crawler.stats.get_stats().items()
+            if k.startswith("request_backout_seconds/")
+        }
+        assert expected_stats == actual_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()
+                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):
+                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())
+
+        matching_log_count = 0
+        for log_record in self.caplog.records:
+            if (
+                str(log_record.msg).startswith("The active response size")
+                and log_record.levelname == "INFO"
+            ):
+                matching_log_count += 1
+        assert matching_log_count == 1
+
+        expected_stats = {
+            "request_backout_seconds/response_max_active_size": gt(0),
+            "request_backout_seconds/total": gt(0),
+        }
+        actual_stats = {
+            k: v
+            for k, v in crawler.stats.get_stats().items()
+            if k.startswith("request_backout_seconds/")
+        }
+        assert expected_stats == actual_stats

tests/test_engine.py

@@ -26,11 +26,7 @@ from scrapy.item import Field, Item
 from scrapy.linkextractors import LinkExtractor
 from scrapy.spiders import Spider
 from scrapy.statscollectors import MemoryStatsCollector
-from scrapy.utils.defer import (
-    _schedule_coro,
-    deferred_from_coro,
-    maybe_deferred_to_future,
-)
+from scrapy.utils.defer import deferred_from_coro, maybe_deferred_to_future
 from scrapy.utils.signal import disconnect_all
 from scrapy.utils.spider import DefaultSpider
 from scrapy.utils.test import get_crawler
@@ -448,10 +444,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,191 @@
+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):
+                with pytest.warns(ScrapyDeprecationWarning):
+                    outcome["active_size"] = (
+                        self.crawler.engine.scraper.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())
+
+        with pytest.warns(ScrapyDeprecationWarning):
+            expected = crawler.engine.scraper.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. Read "
+            "scrapy.core.downloader.DownloaderMiddlewareManager.response_active_size "
+            "instead."
+        )
+
+    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. "
+            "scrapy.core.downloader.DownloaderMiddlewareManager.response_active_size "
+            "might work as a replacement, but modifying that attribute "
+            "might not be a good idea. If you have a use case for it, you "
+            "might want to bring it up in a GitHub issue, to discuss with "
+            "Scrapy developers if there is a better approach, or some "
+            "change we could implement in Scrapy to improve support for "
+            "your use case."
+        )
+
+    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."
+        )

tox.ini

@@ -45,6 +45,7 @@ deps =
     pytest-xdist
     sybil >= 1.3.0  # https://github.com/cjw296/sybil/issues/20#issuecomment-605433422
     testfixtures
+    pywin32; sys_platform == "win32"
     pytest-twisted >= 1.14.3
 
 [testenv]