Make HttpCacheMiddleware wait for in-flight downloads of the same resource

docs/topics/downloader-middleware.rst

@@ -401,6 +401,12 @@ HttpCacheMiddleware
     This middleware provides low-level cache to all HTTP requests and responses.
     It has to be combined with a cache storage backend as well as a cache policy.
 
+    .. versionchanged:: VERSION
+       A request for a resource that another request is already downloading,
+       which requires :attr:`~scrapy.Request.dont_filter` or a custom
+       :setting:`DUPEFILTER_CLASS`, waits for that download to finish and
+       reads its response from the cache.
+
     Scrapy ships with the following HTTP cache storage backends:
 
         * :ref:`httpcache-storage-fs`

docs/topics/stats.rst

@@ -401,6 +401,16 @@ one per actual value of the placeholder.
 
     Set by :class:`~scrapy.downloadermiddlewares.httpcache.HttpCacheMiddleware`.
 
+.. stat:: httpcache/wait
+
+``httpcache/wait``
+    Number of requests that waited for a concurrent request for the same
+    resource to finish downloading.
+
+    .. versionadded:: VERSION
+
+    Set by :class:`~scrapy.downloadermiddlewares.httpcache.HttpCacheMiddleware`.
+
 .. stat:: httpcompression/response_bytes
 
 ``httpcompression/response_bytes``

scrapy/downloadermiddlewares/httpcache.py

@@ -2,8 +2,10 @@ from __future__ import annotations
 
 import logging
 from email.utils import formatdate
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
+from weakref import finalize
 
+from twisted.internet.defer import Deferred
 from twisted.internet.error import ConnectError, ConnectionDone, ConnectionLost
 
 from scrapy import signals
@@ -15,6 +17,7 @@ from scrapy.exceptions import (
     NotConfigured,
 )
 from scrapy.utils.decorators import _warn_spider_arg
+from scrapy.utils.defer import maybe_deferred_to_future
 from scrapy.utils.misc import load_object
 
 if TYPE_CHECKING:
@@ -27,6 +30,7 @@ if TYPE_CHECKING:
     from scrapy.settings import Settings
     from scrapy.spiders import Spider
     from scrapy.statscollectors import StatsCollector
+    from scrapy.utils.request import RequestFingerprinterProtocol
 
 
 logger = logging.getLogger(__name__)
@@ -44,6 +48,7 @@ class HttpCacheMiddleware:
     )
 
     crawler: Crawler
+    _fingerprinter: RequestFingerprinterProtocol
 
     def __init__(self, settings: Settings, stats: StatsCollector) -> None:
         if not settings.getbool("HTTPCACHE_ENABLED"):
@@ -52,6 +57,7 @@ class HttpCacheMiddleware:
         self.storage = load_object(settings["HTTPCACHE_STORAGE"])(settings)
         self.ignore_missing = settings.getbool("HTTPCACHE_IGNORE_MISSING")
         self.stats = stats
+        self._downloading: dict[bytes, list[Deferred[None]]] = {}
 
     @classmethod
     def from_crawler(cls, crawler: Crawler) -> Self:
@@ -60,16 +66,20 @@ class HttpCacheMiddleware:
         crawler.signals.connect(o.spider_opened, signal=signals.spider_opened)
         crawler.signals.connect(o.spider_closed, signal=signals.spider_closed)
         o.crawler = crawler
+        assert crawler.request_fingerprinter
+        o._fingerprinter = crawler.request_fingerprinter
         return o
 
     def spider_opened(self, spider: Spider) -> None:
         self.storage.open_spider(spider)
 
     def spider_closed(self, spider: Spider) -> None:
+        for fingerprint in list(self._downloading):
+            self._wake(fingerprint)
         self.storage.close_spider(spider)
 
     @_warn_spider_arg
-    def process_request(
+    async def process_request(
         self, request: Request, spider: Spider | None = None
     ) -> Request | Response | None:
         if request.meta.get("dont_cache", False):
@@ -81,25 +91,16 @@ class HttpCacheMiddleware:
             return None
 
         # Look for cached response and check if expired
-        cachedresponse: Response | None
-        try:
-            cachedresponse = self.storage.retrieve_response(
-                self.crawler.spider, request
-            )
-        except Exception:
-            self.stats.inc_value("httpcache/retrieve_error")
-            logger.warning(
-                f"Could not read the cache entry for {request}, treating it as a "
-                f"cache miss.",
-                exc_info=True,
-                extra={"spider": self.crawler.spider},
-            )
-            cachedresponse = None
+        cachedresponse = self._retrieve(request)
+        if cachedresponse is None and await self._wait_for_download(request):
+            cachedresponse = self._retrieve(request)
+
         if cachedresponse is None:
             self.stats.inc_value("httpcache/miss")
             if self.ignore_missing:
                 self.stats.inc_value("httpcache/ignore")
                 raise IgnoreRequest(f"Ignored request not in cache: {request}")
+            self._mark_downloading(request)
             return None  # first time request
 
         # Return cached response only if not expired
@@ -111,13 +112,81 @@ class HttpCacheMiddleware:
         # Keep a reference to cached response to avoid a second cache lookup on
         # process_response hook
         request.meta["cached_response"] = cachedresponse
+        self._mark_downloading(request)
 
         return None
 
+    def _retrieve(self, request: Request) -> Response | None:
+        try:
+            return cast(
+                "Response | None",
+                self.storage.retrieve_response(self.crawler.spider, request),
+            )
+        except Exception:
+            self.stats.inc_value("httpcache/retrieve_error")
+            logger.warning(
+                f"Could not read the cache entry for {request}, treating it as a "
+                f"cache miss.",
+                exc_info=True,
+                extra={"spider": self.crawler.spider},
+            )
+            return None
+
+    async def _wait_for_download(self, request: Request) -> bool:
+        """Block until an ongoing download of the same resource finishes, and
+        return whether such a download was found."""
+        fingerprint = self._fingerprinter.fingerprint(request)
+        waiters = self._downloading.get(fingerprint)
+        # A request already marked as downloading is the one being awaited on,
+        # so making it wait would block it forever.
+        if waiters is None or request.meta.get("_httpcache_downloading") == fingerprint:
+            return False
+        self.stats.inc_value("httpcache/wait")
+        waiter: Deferred[None] = Deferred(waiters.remove)
+        waiters.append(waiter)
+        await maybe_deferred_to_future(waiter)
+        return True
+
+    def _mark_downloading(self, request: Request) -> None:
+        fingerprint = self._fingerprinter.fingerprint(request)
+        if fingerprint in self._downloading:
+            return
+        waiters: list[Deferred[None]] = []
+        self._downloading[fingerprint] = waiters
+        request.meta["_httpcache_downloading"] = fingerprint
+        # Some outcomes reach neither process_response() nor
+        # process_exception(), e.g. a middleware with a higher priority raising
+        # from process_response(). Waking waiters up once the request object is
+        # gone keeps them from blocking forever in those cases.
+        finalize(request, self._wake, fingerprint, waiters)
+
+    def _wake(
+        self, fingerprint: bytes, waiters: list[Deferred[None]] | None = None
+    ) -> None:
+        current = self._downloading.get(fingerprint)
+        if current is None or (waiters is not None and current is not waiters):
+            return
+        del self._downloading[fingerprint]
+        while current:
+            current.pop(0).callback(None)
+
+    def _downloaded(self, request: Request) -> None:
+        fingerprint = request.meta.pop("_httpcache_downloading", None)
+        if fingerprint is not None:
+            self._wake(fingerprint)
+
     @_warn_spider_arg
     def process_response(
         self, request: Request, response: Response, spider: Spider | None = None
     ) -> Request | Response:
+        # Waiters are woken up only once the response has been cached, so that
+        # they can read it from the cache.
+        try:
+            return self._process_response(request, response)
+        finally:
+            self._downloaded(request)
+
+    def _process_response(self, request: Request, response: Response) -> Response:
         if request.meta.get("dont_cache", False):
             return response
 
@@ -150,6 +219,7 @@ class HttpCacheMiddleware:
     def process_exception(
         self, request: Request, exception: Exception, spider: Spider | None = None
     ) -> Request | Response | None:
+        self._downloaded(request)
         cachedresponse: Response | None = request.meta.pop("cached_response", None)
         if cachedresponse is not None and isinstance(
             exception, self.DOWNLOAD_EXCEPTIONS

tests/test_downloadermiddleware_httpcache.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import email.utils
+import gc
 import logging
 import shutil
 import tempfile
@@ -11,16 +12,25 @@ from typing import TYPE_CHECKING, Any
 from unittest import mock
 
 import pytest
+from twisted.internet.defer import Deferred
 
+from scrapy import signals
+from scrapy.core.downloader.middleware import DownloaderMiddlewareManager
 from scrapy.downloadermiddlewares.httpcache import HttpCacheMiddleware
 from scrapy.exceptions import IgnoreRequest
 from scrapy.extensions.httpcache import DummyPolicy
 from scrapy.http import HtmlResponse, Request, Response
 from scrapy.spiders import Spider
+from scrapy.utils.defer import (
+    _defer_sleep_async,
+    deferred_from_coro,
+    maybe_deferred_to_future,
+)
 from scrapy.utils.test import get_crawler
+from tests.utils.decorators import coroutine_test
 
 if TYPE_CHECKING:
-    from collections.abc import Generator
+    from collections.abc import Callable, Generator
 
     from scrapy.crawler import Crawler
 
@@ -132,7 +142,8 @@ class StorageTestMixin(TestBase):
             with mock.patch("scrapy.extensions.httpcache.time", return_value=future):
                 assert storage.retrieve_response(crawler.spider, self.request)
 
-    def test_corrupted_cache_entry_is_a_miss(self, caplog):
+    @coroutine_test
+    async def test_corrupted_cache_entry_is_a_miss(self, caplog):
         with self._middleware() as mw:
             spider = mw.crawler.spider
             assert spider
@@ -142,7 +153,7 @@ class StorageTestMixin(TestBase):
 
             caplog.clear()
             with caplog.at_level(logging.WARNING):
-                assert mw.process_request(self.request) is None
+                assert await mw.process_request(self.request) is None
 
             assert "treating it as a cache miss" in caplog.text
             assert mw.crawler.stats.get_value("httpcache/retrieve_error") == 1
@@ -154,7 +165,8 @@ class StorageTestMixin(TestBase):
                 self.response, mw.storage.retrieve_response(spider, self.request)
             )
 
-    def test_corrupted_cache_entry_ignore_missing(self):
+    @coroutine_test
+    async def test_corrupted_cache_entry_ignore_missing(self):
         with self._middleware(HTTPCACHE_IGNORE_MISSING=True) as mw:
             spider = mw.crawler.spider
             assert spider
@@ -163,7 +175,7 @@ class StorageTestMixin(TestBase):
             self._corrupt_cache_entry(mw.storage, spider, self.request)
 
             with pytest.raises(IgnoreRequest):
-                mw.process_request(self.request)
+                await mw.process_request(self.request)
 
             assert mw.crawler.stats.get_value("httpcache/retrieve_error") == 1
             assert mw.crawler.stats.get_value("httpcache/ignore") == 1
@@ -187,10 +199,11 @@ class StorageTestMixin(TestBase):
 class PolicyTestMixin(TestBase):
     """Mixin containing policy-specific test methods."""
 
-    def test_dont_cache(self):
+    @coroutine_test
+    async def test_dont_cache(self):
         with self._middleware() as mw:
             self.request.meta["dont_cache"] = True
-            assert mw.process_request(self.request) is None
+            assert await mw.process_request(self.request) is None
             mw.process_response(self.request, self.response)
             assert mw.storage.retrieve_response(mw.crawler.spider, self.request) is None
 
@@ -207,44 +220,48 @@ class PolicyTestMixin(TestBase):
 class DummyPolicyTestMixin(PolicyTestMixin):
     """Mixin containing dummy policy specific test methods."""
 
-    def test_middleware(self):
+    @coroutine_test
+    async def test_middleware(self):
         with self._middleware() as mw:
-            assert mw.process_request(self.request) is None
+            assert await mw.process_request(self.request) is None
             mw.process_response(self.request, self.response)
-            response = mw.process_request(self.request)
+            response = await mw.process_request(self.request)
             assert isinstance(response, HtmlResponse)
             self.assertEqualResponse(self.response, response)
             assert "cached" in response.flags
 
-    def test_different_request_response_urls(self):
+    @coroutine_test
+    async def test_different_request_response_urls(self):
         with self._middleware() as mw:
             req = Request("http://host.com/path")
             res = Response("http://host2.net/test.html")
-            assert mw.process_request(req) is None
+            assert await mw.process_request(req) is None
             mw.process_response(req, res)
-            cached = mw.process_request(req)
+            cached = await mw.process_request(req)
             assert isinstance(cached, Response)
             self.assertEqualResponse(res, cached)
             assert "cached" in cached.flags
 
-    def test_middleware_ignore_missing(self):
+    @coroutine_test
+    async def test_middleware_ignore_missing(self):
         with self._middleware(HTTPCACHE_IGNORE_MISSING=True) as mw:
             with pytest.raises(IgnoreRequest):
-                mw.process_request(self.request)
+                await mw.process_request(self.request)
             mw.process_response(self.request, self.response)
-            response = mw.process_request(self.request)
+            response = await mw.process_request(self.request)
             assert isinstance(response, HtmlResponse)
             self.assertEqualResponse(self.response, response)
             assert "cached" in response.flags
 
-    def test_middleware_ignore_schemes(self):
+    @coroutine_test
+    async def test_middleware_ignore_schemes(self):
         # http responses are cached by default
         req, res = Request("http://test.com/"), Response("http://test.com/")
         with self._middleware() as mw:
-            assert mw.process_request(req) is None
+            assert await mw.process_request(req) is None
             mw.process_response(req, res)
 
-            cached = mw.process_request(req)
+            cached = await mw.process_request(req)
             assert isinstance(cached, Response), type(cached)
             self.assertEqualResponse(res, cached)
             assert "cached" in cached.flags
@@ -252,19 +269,19 @@ class DummyPolicyTestMixin(PolicyTestMixin):
         # file response is not cached by default
         req, res = Request("file:///tmp/t.txt"), Response("file:///tmp/t.txt")
         with self._middleware() as mw:
-            assert mw.process_request(req) is None
+            assert await mw.process_request(req) is None
             mw.process_response(req, res)
 
             assert mw.storage.retrieve_response(mw.crawler.spider, req) is None
-            assert mw.process_request(req) is None
+            assert await mw.process_request(req) is None
 
         # s3 scheme response is cached by default
         req, res = Request("s3://bucket/key"), Response("s3://bucket/key")
         with self._middleware() as mw:
-            assert mw.process_request(req) is None
+            assert await mw.process_request(req) is None
             mw.process_response(req, res)
 
-            cached = mw.process_request(req)
+            cached = await mw.process_request(req)
             assert isinstance(cached, Response), type(cached)
             self.assertEqualResponse(res, cached)
             assert "cached" in cached.flags
@@ -272,38 +289,40 @@ class DummyPolicyTestMixin(PolicyTestMixin):
         # ignore s3 scheme
         req, res = Request("s3://bucket/key2"), Response("s3://bucket/key2")
         with self._middleware(HTTPCACHE_IGNORE_SCHEMES=["s3"]) as mw:
-            assert mw.process_request(req) is None
+            assert await mw.process_request(req) is None
             mw.process_response(req, res)
 
             assert mw.storage.retrieve_response(mw.crawler.spider, req) is None
-            assert mw.process_request(req) is None
+            assert await mw.process_request(req) is None
 
-    def test_middleware_ignore_http_codes(self):
+    @coroutine_test
+    async def test_middleware_ignore_http_codes(self):
         # test response is not cached
         with self._middleware(HTTPCACHE_IGNORE_HTTP_CODES=[202]) as mw:
-            assert mw.process_request(self.request) is None
+            assert await mw.process_request(self.request) is None
             mw.process_response(self.request, self.response)
 
             assert mw.storage.retrieve_response(mw.crawler.spider, self.request) is None
-            assert mw.process_request(self.request) is None
+            assert await mw.process_request(self.request) is None
 
         # test response is cached
         with self._middleware(HTTPCACHE_IGNORE_HTTP_CODES=[203]) as mw:
             mw.process_response(self.request, self.response)
-            response = mw.process_request(self.request)
+            response = await mw.process_request(self.request)
             assert isinstance(response, HtmlResponse)
             self.assertEqualResponse(self.response, response)
             assert "cached" in response.flags
 
-    def test_revalidation_keeps_cached_response(self):
+    @coroutine_test
+    async def test_revalidation_keeps_cached_response(self):
         # The dummy policy considers every cached response valid, so a policy
         # that subclasses it to force revalidation always gets the cached
         # response back, whatever the new response is.
         with self._middleware(HTTPCACHE_POLICY=AlwaysStalePolicy) as mw:
-            assert mw.process_request(self.request) is None
+            assert await mw.process_request(self.request) is None
             mw.process_response(self.request, self.response)
 
-            assert mw.process_request(self.request) is None
+            assert await mw.process_request(self.request) is None
             fresh_response = self.response.replace(body=b"new body")
             response = mw.process_response(self.request, fresh_response)
             assert isinstance(response, Response)
@@ -316,12 +335,12 @@ class RFC2616PolicyTestMixin(PolicyTestMixin):
     """Mixin containing RFC2616 policy specific test methods."""
 
     @staticmethod
-    def _process_requestresponse(
+    async def _process_requestresponse(
         mw: HttpCacheMiddleware, request: Request, response: Response | None
     ) -> Response:
         result: Request | Response | None = None
         try:
-            result = mw.process_request(request)
+            result = await mw.process_request(request)
             if result:
                 assert isinstance(result, Response)
                 return result
@@ -335,7 +354,8 @@ class RFC2616PolicyTestMixin(PolicyTestMixin):
             print("Result", result)
             raise
 
-    def test_request_cacheability(self):
+    @coroutine_test
+    async def test_request_cacheability(self):
         res0 = Response(
             self.request.url, status=200, headers={"Expires": self.tomorrow}
         )
@@ -344,27 +364,28 @@ class RFC2616PolicyTestMixin(PolicyTestMixin):
         req2 = req0.replace(headers={"Cache-Control": "no-cache"})
         with self._middleware() as mw:
             # response for a request with no-store must not be cached
-            res1 = self._process_requestresponse(mw, req1, res0)
+            res1 = await self._process_requestresponse(mw, req1, res0)
             self.assertEqualResponse(res1, res0)
             assert mw.storage.retrieve_response(mw.crawler.spider, req1) is None
             # Re-do request without no-store and expect it to be cached
-            res2 = self._process_requestresponse(mw, req0, res0)
+            res2 = await self._process_requestresponse(mw, req0, res0)
             assert "cached" not in res2.flags
-            res3 = mw.process_request(req0)
+            res3 = await mw.process_request(req0)
             assert isinstance(res3, Response)
             assert "cached" in res3.flags
             self.assertEqualResponse(res2, res3)
             # request with no-cache directive must not return cached response
             # but it allows new response to be stored
             res0b = res0.replace(body=b"foo")
-            res4 = self._process_requestresponse(mw, req2, res0b)
+            res4 = await self._process_requestresponse(mw, req2, res0b)
             self.assertEqualResponse(res4, res0b)
             assert "cached" not in res4.flags
-            res5 = self._process_requestresponse(mw, req0, None)
+            res5 = await self._process_requestresponse(mw, req0, None)
             self.assertEqualResponse(res5, res0b)
             assert "cached" in res5.flags
 
-    def test_response_cacheability(self):
+    @coroutine_test
+    async def test_response_cacheability(self):
         responses = [
             # 304 is not cacheable no matter what servers sends
             (False, 304, {}),
@@ -400,9 +421,9 @@ class RFC2616PolicyTestMixin(PolicyTestMixin):
             for idx, (shouldcache, status, headers) in enumerate(responses):
                 req0 = Request(f"http://example-{idx}.com")
                 res0 = Response(req0.url, status=status, headers=headers)
-                res1 = self._process_requestresponse(mw, req0, res0)
+                res1 = await self._process_requestresponse(mw, req0, res0)
                 res304 = res0.replace(status=304)
-                res2 = self._process_requestresponse(
+                res2 = await self._process_requestresponse(
                     mw, req0, res304 if shouldcache else res0
                 )
                 self.assertEqualResponse(res1, res0)
@@ -424,9 +445,9 @@ class RFC2616PolicyTestMixin(PolicyTestMixin):
                 )
                 req0 = Request(f"http://example2-{idx}.com")
                 res0 = Response(req0.url, status=status, headers=headers)
-                res1 = self._process_requestresponse(mw, req0, res0)
+                res1 = await self._process_requestresponse(mw, req0, res0)
                 res304 = res0.replace(status=304)
-                res2 = self._process_requestresponse(
+                res2 = await self._process_requestresponse(
                     mw, req0, res304 if shouldcache else res0
                 )
                 self.assertEqualResponse(res1, res0)
@@ -440,7 +461,8 @@ class RFC2616PolicyTestMixin(PolicyTestMixin):
                     assert not resc
                     assert "cached" not in res2.flags
 
-    def test_cached_and_fresh(self):
+    @coroutine_test
+    async def test_cached_and_fresh(self):
         sampledata = [
             (200, {"Date": self.yesterday, "Expires": self.tomorrow}),
             (200, {"Date": self.yesterday, "Cache-Control": "max-age=86405"}),
@@ -488,22 +510,23 @@ class RFC2616PolicyTestMixin(PolicyTestMixin):
                 req0 = Request(f"http://example-{idx}.com")
                 res0 = Response(req0.url, status=status, headers=headers)
                 # cache fresh response
-                res1 = self._process_requestresponse(mw, req0, res0)
+                res1 = await self._process_requestresponse(mw, req0, res0)
                 self.assertEqualResponse(res1, res0)
                 assert "cached" not in res1.flags
                 # return fresh cached response without network interaction
-                res2 = self._process_requestresponse(mw, req0, None)
+                res2 = await self._process_requestresponse(mw, req0, None)
                 self.assertEqualResponse(res1, res2)
                 assert "cached" in res2.flags
                 # validate cached response if request max-age set as 0
                 req1 = req0.replace(headers={"Cache-Control": "max-age=0"})
                 res304 = res0.replace(status=304)
-                assert mw.process_request(req1) is None
-                res3 = self._process_requestresponse(mw, req1, res304)
+                assert await mw.process_request(req1) is None
+                res3 = await self._process_requestresponse(mw, req1, res304)
                 self.assertEqualResponse(res1, res3)
                 assert "cached" in res3.flags
 
-    def test_cached_and_stale(self):
+    @coroutine_test
+    async def test_cached_and_stale(self):
         sampledata = [
             (200, {"Date": self.today, "Expires": self.yesterday}),
             (
@@ -545,13 +568,13 @@ class RFC2616PolicyTestMixin(PolicyTestMixin):
                 req0 = Request(f"http://example-{idx}.com")
                 res0a = Response(req0.url, status=status, headers=headers)
                 # cache expired response
-                res1 = self._process_requestresponse(mw, req0, res0a)
+                res1 = await self._process_requestresponse(mw, req0, res0a)
                 self.assertEqualResponse(res1, res0a)
                 assert "cached" not in res1.flags
                 # Same request but as cached response is stale a new response must
                 # be returned
                 res0b = res0a.replace(body=b"bar")
-                res2 = self._process_requestresponse(mw, req0, res0b)
+                res2 = await self._process_requestresponse(mw, req0, res0b)
                 self.assertEqualResponse(res2, res0b)
                 assert "cached" not in res2.flags
                 cc = headers.get("Cache-Control", "")
@@ -560,13 +583,13 @@ class RFC2616PolicyTestMixin(PolicyTestMixin):
                 # are present
                 if "ETag" in headers or "Last-Modified" in headers:
                     res0c = res0b.replace(status=304)
-                    res3 = self._process_requestresponse(mw, req0, res0c)
+                    res3 = await self._process_requestresponse(mw, req0, res0c)
                     self.assertEqualResponse(res3, res0b)
                     assert "cached" in res3.flags
                     # get cached response on server errors unless must-revalidate
                     # in cached response
                     res0d = res0b.replace(status=500)
-                    res4 = self._process_requestresponse(mw, req0, res0d)
+                    res4 = await self._process_requestresponse(mw, req0, res0d)
                     if "must-revalidate" in cc:
                         assert "cached" not in res4.flags
                         self.assertEqualResponse(res4, res0d)
@@ -576,78 +599,84 @@ class RFC2616PolicyTestMixin(PolicyTestMixin):
                 # Requests with max-stale can fetch expired cached responses
                 # unless cached response has must-revalidate
                 req1 = req0.replace(headers={"Cache-Control": "max-stale"})
-                res5 = self._process_requestresponse(mw, req1, res0b)
+                res5 = await self._process_requestresponse(mw, req1, res0b)
                 self.assertEqualResponse(res5, res0b)
                 if "no-cache" in cc or "must-revalidate" in cc:
                     assert "cached" not in res5.flags
                 else:
                     assert "cached" in res5.flags
 
-    def test_middleware_ignore_schemes(self):
+    @coroutine_test
+    async def test_middleware_ignore_schemes(self):
         # file responses are not cached by default
         req = Request("file:///tmp/t.txt")
         res = Response(req.url, headers={"Expires": self.tomorrow})
         with self._middleware() as mw:
-            assert mw.process_request(req) is None
+            assert await mw.process_request(req) is None
             mw.process_response(req, res)
 
             assert mw.storage.retrieve_response(mw.crawler.spider, req) is None
-            assert mw.process_request(req) is None
+            assert await mw.process_request(req) is None
 
-    def test_max_stale_with_value(self):
+    @coroutine_test
+    async def test_max_stale_with_value(self):
         # A response that expired one day ago.
         headers = {"Date": self.yesterday, "Expires": self.yesterday}
         with self._middleware() as mw:
             req0 = Request("http://example.com")
             res0 = Response(req0.url, headers=headers)
-            self._process_requestresponse(mw, req0, res0)
+            await self._process_requestresponse(mw, req0, res0)
 
             # max-stale greater than the staleness of the cached response
             req1 = req0.replace(headers={"Cache-Control": "max-stale=172800"})
-            res1 = mw.process_request(req1)
+            res1 = await mw.process_request(req1)
             assert isinstance(res1, Response)
             assert "cached" in res1.flags
 
             # max-stale lower than the staleness of the cached response
             req2 = req0.replace(headers={"Cache-Control": "max-stale=60"})
-            assert mw.process_request(req2) is None
+            assert await mw.process_request(req2) is None
+            mw.process_exception(req2, IgnoreRequest())
 
             # a non-integer max-stale value is ignored
             req3 = req0.replace(headers={"Cache-Control": "max-stale=soon"})
-            assert mw.process_request(req3) is None
+            assert await mw.process_request(req3) is None
 
-    def test_response_dated_in_the_future(self):
+    @coroutine_test
+    async def test_response_dated_in_the_future(self):
         # A Date header ahead of the local clock must not make the cached
         # response look aged.
         headers = {"Date": self.tomorrow, "Cache-Control": "max-age=10"}
         with self._middleware() as mw:
             req0 = Request("http://example.com")
             res0 = Response(req0.url, headers=headers)
-            res1 = self._process_requestresponse(mw, req0, res0)
+            res1 = await self._process_requestresponse(mw, req0, res0)
             assert "cached" not in res1.flags
 
-            res2 = self._process_requestresponse(mw, req0, None)
+            res2 = await self._process_requestresponse(mw, req0, None)
             self.assertEqualResponse(res1, res2)
             assert "cached" in res2.flags
 
-    def test_process_exception(self):
+    @coroutine_test
+    async def test_process_exception(self):
         with self._middleware() as mw:
             res0 = Response(self.request.url, headers={"Expires": self.yesterday})
             req0 = Request(self.request.url)
-            self._process_requestresponse(mw, req0, res0)
+            await self._process_requestresponse(mw, req0, res0)
             for e in mw.DOWNLOAD_EXCEPTIONS:
                 # Simulate encountering an error on download attempts
-                assert mw.process_request(req0) is None
+                assert await mw.process_request(req0) is None
                 res1 = mw.process_exception(req0, e("foo"))
                 # Use cached response as recovery
                 assert isinstance(res1, Response)
                 assert "cached" in res1.flags
                 self.assertEqualResponse(res0, res1)
             # Do not use cached response for unhandled exceptions
-            mw.process_request(req0)
+            await mw.process_request(req0)
             assert mw.process_exception(req0, Exception("foo")) is None
 
-    def test_ignore_response_cache_controls(self):
+    @coroutine_test
+    async def test_ignore_response_cache_controls(self):
         sampledata = [
             (200, {"Date": self.yesterday, "Expires": self.tomorrow}),
             (200, {"Date": self.yesterday, "Cache-Control": "no-store,max-age=86405"}),
@@ -662,11 +691,11 @@ class RFC2616PolicyTestMixin(PolicyTestMixin):
                 req0 = Request(f"http://example-{idx}.com")
                 res0 = Response(req0.url, status=status, headers=headers)
                 # cache fresh response
-                res1 = self._process_requestresponse(mw, req0, res0)
+                res1 = await self._process_requestresponse(mw, req0, res0)
                 self.assertEqualResponse(res1, res0)
                 assert "cached" not in res1.flags
                 # return fresh cached response without network interaction
-                res2 = self._process_requestresponse(mw, req0, None)
+                res2 = await self._process_requestresponse(mw, req0, None)
                 self.assertEqualResponse(res1, res2)
                 assert "cached" in res2.flags
 
@@ -732,3 +761,279 @@ class TestFilesystemStorageGzipWithDummyPolicy(TestFilesystemStorageWithDummyPol
         # A spider killed while writing a gzip file leaves it truncated.
         body_path = Path(storage._get_request_path(spider, request), "response_body")
         body_path.write_bytes(body_path.read_bytes()[:-5])
+
+
+class _LateMiddleware:
+    """Downloader middleware that runs after :class:`HttpCacheMiddleware`."""
+
+    priority = 950
+
+
+class _IgnoreOnRequest(_LateMiddleware):
+    def __init__(self) -> None:
+        self.gate: Deferred[None] = Deferred()
+
+    async def process_request(self, request: Request) -> None:
+        await maybe_deferred_to_future(self.gate)
+        raise IgnoreRequest
+
+
+class _RaiseOnResponse(_LateMiddleware):
+    def process_response(self, request: Request, response: Response) -> Response:
+        raise ValueError("Late middleware failure")
+
+
+class _RedirectOnResponse(_LateMiddleware):
+    def process_response(self, request: Request, response: Response) -> Request:
+        return Request("http://www.example.com/elsewhere")
+
+
+class _Downloads:
+    """Download function that records its calls and, until released, blocks."""
+
+    def __init__(self, status: int = 200) -> None:
+        self.urls: list[str] = []
+        self.status = status
+        self.blocking = True
+        self._blocked: list[Deferred[None]] = []
+
+    async def __call__(self, request: Request) -> Response:
+        self.urls.append(request.url)
+        if self.blocking:
+            blocked: Deferred[None] = Deferred()
+            self._blocked.append(blocked)
+            await maybe_deferred_to_future(blocked)
+        return Response(request.url, status=self.status, body=b"body")
+
+    def release(self) -> None:
+        self.blocking = False
+        while self._blocked:
+            self._blocked.pop(0).callback(None)
+
+
+class TestConcurrentDownloads(TestBase):
+    """Requests for a resource that a concurrent request is already
+    downloading wait for that download and read its response from the
+    cache."""
+
+    policy_class = "scrapy.extensions.httpcache.DummyPolicy"
+    storage_class = "scrapy.extensions.httpcache.FilesystemCacheStorage"
+    url = "http://www.example.com"
+
+    @contextmanager
+    def _manager(
+        self, *late_middlewares: type[_LateMiddleware], **new_settings: Any
+    ) -> Generator[tuple[DownloaderMiddlewareManager, Crawler]]:
+        new_settings.setdefault(
+            "DOWNLOADER_MIDDLEWARES",
+            {mw: mw.priority for mw in late_middlewares},
+        )
+        with self._get_crawler(**new_settings) as crawler:
+            manager = DownloaderMiddlewareManager.from_crawler(crawler)
+            crawler.signals.send_catch_log(signals.spider_opened, spider=crawler.spider)
+            try:
+                yield manager, crawler
+            finally:
+                crawler.signals.send_catch_log(
+                    signals.spider_closed, spider=crawler.spider, reason="finished"
+                )
+
+    @staticmethod
+    def _middleware_of(manager: DownloaderMiddlewareManager, base: type) -> Any:
+        return next(mw for mw in manager.middlewares if isinstance(mw, base))
+
+    @staticmethod
+    async def _wait_until(condition: Callable[[], Any]) -> None:
+        for _ in range(100):
+            if condition():
+                return
+            await _defer_sleep_async()
+        raise AssertionError("The expected state was never reached")
+
+    async def _start(
+        self,
+        manager: DownloaderMiddlewareManager,
+        download: _Downloads,
+        crawler: Crawler,
+    ) -> tuple[Request, Deferred[Any], Deferred[Any]]:
+        """Start two concurrent requests for the same URL, the second one
+        waiting for the first one.
+
+        Return the first request, which callers must keep a reference to for
+        as long as they do not want it garbage-collected, and the results of
+        both requests.
+        """
+        request = Request(self.url)
+        first = deferred_from_coro(manager.download_async(download, request))
+        await self._wait_until(lambda: download.urls)
+        second = deferred_from_coro(manager.download_async(download, Request(self.url)))
+        stats = crawler.stats
+        assert stats
+        await self._wait_until(lambda: stats.get_value("httpcache/wait"))
+        return request, first, second
+
+    @coroutine_test
+    async def test_cached(self):
+        with self._manager() as (manager, crawler):
+            download = _Downloads()
+            _request, first, second = await self._start(manager, download, crawler)
+            download.release()
+
+            assert isinstance(await maybe_deferred_to_future(first), Response)
+            response = await maybe_deferred_to_future(second)
+            assert isinstance(response, Response)
+            assert "cached" in response.flags
+            assert response.body == b"body"
+            assert download.urls == [self.url]
+
+    @coroutine_test
+    async def test_uncacheable_response(self):
+        with self._manager(HTTPCACHE_IGNORE_HTTP_CODES=[404]) as (manager, crawler):
+            download = _Downloads(status=404)
+            _request, first, second = await self._start(manager, download, crawler)
+            download.release()
+
+            for result in (first, second):
+                response = await maybe_deferred_to_future(result)
+                assert isinstance(response, Response)
+                assert "cached" not in response.flags
+            assert download.urls == [self.url] * 2
+
+    @coroutine_test
+    async def test_download_exception(self):
+        with self._manager() as (manager, crawler):
+            download = _Downloads()
+            _request, first, second = await self._start(manager, download, crawler)
+            download._blocked.pop(0).errback(ValueError("Download failure"))
+
+            with pytest.raises(ValueError, match="Download failure"):
+                await maybe_deferred_to_future(first)
+            download.release()
+            assert isinstance(await maybe_deferred_to_future(second), Response)
+            assert download.urls == [self.url] * 2
+
+    @coroutine_test
+    async def test_ignored_by_a_later_middleware(self):
+        with self._manager(_IgnoreOnRequest) as (manager, crawler):
+            download = _Downloads()
+            late = self._middleware_of(manager, _IgnoreOnRequest)
+            middleware = self._middleware_of(manager, HttpCacheMiddleware)
+            request = Request(self.url)
+            first = deferred_from_coro(manager.download_async(download, request))
+            await self._wait_until(lambda: middleware._downloading)
+            second = deferred_from_coro(
+                manager.download_async(download, Request(self.url))
+            )
+            stats = crawler.stats
+            assert stats
+            await self._wait_until(lambda: stats.get_value("httpcache/wait"))
+            late.gate.callback(None)
+
+            for result in (first, second):
+                with pytest.raises(IgnoreRequest):
+                    await maybe_deferred_to_future(result)
+            assert not download.urls
+
+    @coroutine_test
+    async def test_response_replaced_by_a_later_middleware(self):
+        # A middleware with a higher priority returning a request from
+        # process_response() keeps HttpCacheMiddleware.process_response() from
+        # running, so waiters are only woken up once the first request object
+        # is garbage-collected.
+        with self._manager(_RedirectOnResponse) as (manager, crawler):
+            download = _Downloads()
+            request, first, second = await self._start(manager, download, crawler)
+            middleware = self._middleware_of(manager, HttpCacheMiddleware)
+            download.release()
+
+            assert isinstance(await maybe_deferred_to_future(first), Request)
+            assert middleware._downloading
+
+            def collected() -> bool:
+                gc.collect()
+                return not middleware._downloading
+
+            del request
+            await self._wait_until(collected)
+            assert isinstance(await maybe_deferred_to_future(second), Request)
+            assert download.urls == [self.url] * 2
+
+    @coroutine_test
+    async def test_response_error_in_a_later_middleware(self):
+        # A middleware with a higher priority raising from process_response()
+        # also keeps HttpCacheMiddleware.process_response() from running, but
+        # here the resulting failure keeps the first request object alive, so
+        # waiters block until the spider is closed.
+        with self._manager(_RaiseOnResponse) as (manager, crawler):
+            download = _Downloads()
+            request, first, second = await self._start(manager, download, crawler)
+            middleware = self._middleware_of(manager, HttpCacheMiddleware)
+            download.release()
+
+            with pytest.raises(ValueError, match="Late middleware failure"):
+                await maybe_deferred_to_future(first)
+            del request
+            gc.collect()
+            assert middleware._downloading
+            assert not second.called
+
+            middleware.spider_closed(crawler.spider)
+            with pytest.raises(ValueError, match="Late middleware failure"):
+                await maybe_deferred_to_future(second)
+            assert download.urls == [self.url] * 2
+
+    @coroutine_test
+    async def test_cancelled_waiter(self):
+        with self._manager() as (manager, crawler):
+            download = _Downloads()
+            _request, first, second = await self._start(manager, download, crawler)
+            second.cancel()
+            download.release()
+
+            assert isinstance(await maybe_deferred_to_future(first), Response)
+            assert download.urls == [self.url]
+            assert not self._middleware_of(manager, HttpCacheMiddleware)._downloading
+
+
+class TestInFlightTracking(TestBase):
+    """Bookkeeping of requests that are being downloaded."""
+
+    policy_class = "scrapy.extensions.httpcache.DummyPolicy"
+    storage_class = "scrapy.extensions.httpcache.FilesystemCacheStorage"
+    url = "http://www.example.com"
+
+    @coroutine_test
+    async def test_garbage_collected_request(self):
+        with self._middleware() as mw:
+            request = Request(self.url)
+            assert await mw.process_request(request) is None
+            assert mw._downloading
+
+            del request
+            gc.collect()
+            assert not mw._downloading
+
+    @coroutine_test
+    async def test_same_request_twice(self):
+        with self._middleware() as mw:
+            request = Request(self.url)
+            assert await mw.process_request(request) is None
+            assert await mw.process_request(request) is None
+
+    @coroutine_test
+    async def test_spider_closed(self):
+        with self._middleware() as mw:
+            request = Request(self.url)
+            assert await mw.process_request(request) is None
+            waiting = deferred_from_coro(mw.process_request(Request(self.url)))
+            assert mw.crawler.stats
+            for _ in range(100):
+                if mw.crawler.stats.get_value("httpcache/wait"):
+                    break
+                await _defer_sleep_async()
+            assert not waiting.called
+
+            assert mw.crawler.spider
+            mw.spider_closed(mw.crawler.spider)
+            assert await maybe_deferred_to_future(waiting) is None
+            assert not mw._downloading