Implement single-request delay support

docs/topics/request-response.rst

@@ -876,6 +876,8 @@ Those are:
 * :reqmeta:`autothrottle_dont_adjust_delay`
 * :reqmeta:`bindaddress`
 * :reqmeta:`cookiejar`
+* :reqmeta:`delay`
+* :reqmeta:`delayed`
 * :reqmeta:`dont_cache`
 * :reqmeta:`dont_merge_cookies`
 * :reqmeta:`dont_obey_robotstxt`
@@ -943,6 +945,44 @@ This meta key is not supported by
 :class:`~scrapy.core.downloader.handlers._httpx.HttpxDownloadHandler`, but the
 :setting:`DOWNLOAD_BIND_ADDRESS` is supported by it.
 
+.. reqmeta:: delay
+
+delay
+-----
+
+.. versionadded:: VERSION
+
+Number of seconds to hold a request before sending it:
+
+.. code-block:: python
+
+    Request("https://example.com/slow", meta={"delay": 5.0})
+
+The delay is a minimum: once it elapses, the request competes with the other
+pending requests as usual, subject to e.g. its
+:attr:`~scrapy.Request.priority`, :setting:`CONCURRENT_REQUESTS_PER_DOMAIN` and
+:setting:`DOWNLOAD_DELAY`.
+
+Scrapy removes this key as soon as the countdown starts, so that a copy of the
+request, e.g. a retry or a redirect, is not held again, and so that setting the
+key again asks for a new delay. See also :reqmeta:`delayed`.
+
+.. reqmeta:: delayed
+
+delayed
+-------
+
+.. versionadded:: VERSION
+
+Delays, in seconds, that a request has been held for because of
+:reqmeta:`delay`. Read-only.
+
+.. code-block:: python
+
+    def parse(self, response):
+        delays = response.meta.get("delayed", [])
+        self.logger.info(f"Held {len(delays)} times, {sum(delays)}s in total")
+
 .. reqmeta:: download_timeout
 
 download_timeout

scrapy/core/downloader/__init__.py

@@ -4,7 +4,7 @@ import random
 from collections import deque
 from dataclasses import dataclass, field
 from datetime import datetime
-from time import monotonic
+from time import monotonic, time
 from typing import TYPE_CHECKING, Any
 
 from twisted.internet.defer import Deferred, inlineCallbacks
@@ -19,6 +19,7 @@ from scrapy.utils.asyncio import (
     CallLaterResult,
     call_later,
     create_looping_call,
+    sleep,
 )
 from scrapy.utils.decorators import _warn_spider_arg
 from scrapy.utils.defer import (
@@ -40,6 +41,30 @@ if TYPE_CHECKING:
     from scrapy.signalmanager import SignalManager
 
 
+# Request.meta key holding the delay of a request and the time before which it
+# must not be sent because of it. The deadline is a wall-clock time, so that it
+# survives a crawl being resumed from a disk queue.
+_DELAY_DEADLINE_META_KEY = "_delay_deadline"
+
+
+def _start_delay_countdown(request: Request) -> None:
+    """Turn the ``delay`` meta key of *request*, if any, into a deadline.
+
+    Requests are stamped as they are scheduled, so that their delay runs while
+    they wait for their turn instead of only once they get it.
+
+    ``delay`` is consumed, so that a copy of the request, such as a retry or a
+    redirect, is not held again, while setting the key again asks for a new
+    delay. What the request has been through is left on ``delayed``.
+    """
+    delay = request.meta.pop("delay", None)
+    if not delay:
+        return
+    delay = float(delay)
+    request.meta[_DELAY_DEADLINE_META_KEY] = (delay, time() + delay)
+    request.meta["delayed"] = [*request.meta.get("delayed", []), delay]
+
+
 @dataclass(slots=True, eq=False)
 class Slot:
     """Downloader slot"""
@@ -156,8 +181,30 @@ class Downloader:
 
         return key
 
+    async def _delay_request(self, request: Request) -> None:
+        """Hold *request* until the deadline of its ``delay`` meta key.
+
+        The wait happens before the request gets a slot, so it does not take a
+        slot concurrency spot.
+        """
+        # A request that skipped the scheduler, e.g. one from
+        # ExecutionEngine.download_async(), starts its countdown here.
+        _start_delay_countdown(request)
+        countdown = request.meta.pop(_DELAY_DEADLINE_META_KEY, None)
+        if countdown is None:
+            return
+        delay, deadline = countdown
+        # A deadline further away than the delay means the clock moved backwards
+        # since it was set, which takes a crawl resumed from a disk queue, e.g.
+        # on a machine with a lagging clock. Capping keeps the wait to what the
+        # delay asked for.
+        wait = min(deadline - time(), delay)
+        if wait > 0:
+            await sleep(wait)
+
     # passed as download_func into self.middleware.download() in self.fetch()
     async def _enqueue_request(self, request: Request) -> Response:
+        await self._delay_request(request)
         key, slot = self._get_slot(request)
         request.meta[self.DOWNLOAD_SLOT] = key
         slot.active.add(request)

scrapy/core/engine.py

@@ -20,6 +20,7 @@ from twisted.internet.defer import CancelledError, Deferred, inlineCallbacks
 from twisted.python.failure import Failure
 
 from scrapy import signals
+from scrapy.core.downloader import _start_delay_countdown
 from scrapy.core.scheduler import BaseScheduler
 from scrapy.core.scraper import Scraper
 from scrapy.exceptions import (
@@ -438,6 +439,7 @@ class ExecutionEngine:
         self._slot.nextcall.schedule()  # type: ignore[union-attr]
 
     def _schedule_request(self, request: Request) -> None:
+        _start_delay_countdown(request)
         request_scheduled_result = self.signals.send_catch_log(
             signals.request_scheduled,
             request=request,

tests/test_core_downloader.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
 import warnings
-from typing import TYPE_CHECKING, cast
+from time import monotonic, time
+from typing import TYPE_CHECKING, Any, cast
 
 import OpenSSL.SSL
 import pytest
@@ -33,6 +34,7 @@ from scrapy.utils.spider import DefaultSpider
 from scrapy.utils.test import get_crawler
 from tests.mockserver.http_resources import PayloadResource, put_child
 from tests.mockserver.utils import ssl_context_factory
+from tests.spiders import MockServerSpider
 from tests.utils.decorators import coroutine_test
 
 if TYPE_CHECKING:
@@ -41,6 +43,7 @@ if TYPE_CHECKING:
     from twisted.web.iweb import IBodyProducer
 
     from scrapy.http import Response
+    from tests.mockserver.http import MockServer
 
 
 class TestSlot:
@@ -49,6 +52,87 @@ class TestSlot:
         assert repr(slot) == "Slot(concurrency=8, delay=0.1, randomize_delay=True)"
 
 
+def _downloader() -> Downloader:
+    crawler = get_crawler(DefaultSpider, {"DOWNLOADER_MIDDLEWARES_BASE": {}})
+    crawler.spider = crawler._create_spider()
+    return Downloader(crawler)
+
+
+@coroutine_test
+async def test_delay_meta() -> None:
+    delay = 0.2
+    downloader = _downloader()
+    request = Request("data:,", meta={"delay": delay})
+    try:
+        start = monotonic()
+        await maybe_deferred_to_future(downloader.fetch(request))
+        delayed = monotonic() - start
+        await maybe_deferred_to_future(downloader.fetch(request))
+        held_again = monotonic() - start - delayed
+        request.meta["delay"] = delay
+        await maybe_deferred_to_future(downloader.fetch(request))
+        held_for_the_new_delay = monotonic() - start - delayed - held_again
+    finally:
+        downloader.close()
+    assert delayed >= delay
+    assert "delay" not in request.meta
+    # Downloading the same request again does not hold it again, but setting the
+    # key again does.
+    assert held_again < delay
+    assert held_for_the_new_delay >= delay
+    assert request.meta["delayed"] == [delay, delay]
+
+
+@coroutine_test
+async def test_delay_meta_stale_deadline() -> None:
+    delay = 0.2
+    downloader = _downloader()
+    # A deadline from a different run of the crawl, restored from a disk queue,
+    # is capped at the delay.
+    request = Request("data:,", meta={"_delay_deadline": (delay, time() + 10 * delay)})
+    try:
+        start = monotonic()
+        await maybe_deferred_to_future(downloader.fetch(request))
+        elapsed = monotonic() - start
+    finally:
+        downloader.close()
+    assert delay <= elapsed < 2 * delay
+
+
+class DelaySpider(MockServerSpider):
+    name = "delay_meta"
+    custom_settings = {"CONCURRENT_REQUESTS": 1}
+
+    def __init__(self, *args: Any, **kwargs: Any):
+        super().__init__(*args, **kwargs)
+        self.times: list[float] = []
+        self.start_time: float = monotonic()
+
+    async def start(self):
+        assert self.mockserver
+        # The slow request keeps the delayed one waiting for its turn for longer
+        # than its delay.
+        yield Request(self.mockserver.url("/delay?n=1"))
+        yield Request(self.mockserver.url("/status?n=200"), meta={"delay": 0.9})
+
+    def parse(self, response):
+        self.times.append(monotonic() - self.start_time)
+
+
+@coroutine_test
+async def test_delay_meta_countdown_starts_when_scheduled(
+    mockserver: MockServer,
+) -> None:
+    crawler = get_crawler(DelaySpider)
+    await crawler.crawl_async(mockserver=mockserver)
+    spider = crawler.spider
+    assert isinstance(spider, DelaySpider)
+    assert len(spider.times) == 2
+    # The delay ran while the request waited for its turn, so it was sent as
+    # soon as it got one, rather than 0.9s later.
+    assert spider.times[1] < spider.times[0] + 0.5
+
+
 @pytest.mark.requires_reactor  # this test is related to the Twisted HTTP code
 class TestContextFactoryBase:
     @async_yield_fixture