diff --git a/docs/faq.rst b/docs/faq.rst
index 73a72a2df..2406ffcc0 100644
--- a/docs/faq.rst
+++ b/docs/faq.rst
@@ -413,6 +413,23 @@ inline within web pages, similar to external resources. The "data:" scheme with
 content (",") essentially creates a request to a data URL without any specific content.
 
 
+.. _faq-too-many-open-files:
+
+I get ``too many file descriptors in select()`` or ``Too many open files``
+--------------------------------------------------------------------------
+
+The crawl is opening more sockets than the system allows. On Windows
+:func:`select.select` handles at most 512 sockets; elsewhere the ceiling is the
+maximum number of open files allowed for the process, which sockets, log files
+and feed exports all count towards.
+
+Lower :setting:`CONCURRENT_CONNECTIONS_PER_HANDLER`, keeping in mind that every
+:ref:`download handler <topics-download-handlers>` applies it on its own, so a
+crawl of both ``http://`` and ``https://`` URLs can open twice as many
+connections. Raising the limit of the operating system works as well on
+platforms other than Windows.
+
+
 Running ``runspider`` I get ``error: No spider found in file: <filename>``
 --------------------------------------------------------------------------
 
diff --git a/docs/topics/optimize.rst b/docs/topics/optimize.rst
index cea49fca3..3b0c29ff5 100644
--- a/docs/topics/optimize.rst
+++ b/docs/topics/optimize.rst
@@ -132,6 +132,10 @@ Raise them to crawl a single website faster, and see
 :ref:`broad-crawls-concurrency` to spread requests across many websites
 instead.
 
+Concurrency also drives how many connections stay open, which is capped
+separately by :setting:`CONCURRENT_CONNECTIONS_PER_HANDLER`. See
+:ref:`faq-too-many-open-files` if a crawl runs out of file descriptors.
+
 The limit that matters, though, is the one the target website tolerates.
 Exceeding it gets you throttled, served errors or banned, all of which make the
 crawl slower than a lower concurrency would have been. To find that limit:
diff --git a/docs/topics/settings.rst b/docs/topics/settings.rst
index e58815fad..0f9501ed5 100644
--- a/docs/topics/settings.rst
+++ b/docs/topics/settings.rst
@@ -567,6 +567,44 @@ project name). This name will be used for the logging too.
 It's automatically populated with your project name when you create your
 project with the :command:`startproject` command.
 
+.. setting:: CONCURRENT_CONNECTIONS_PER_HANDLER
+
+CONCURRENT_CONNECTIONS_PER_HANDLER
+----------------------------------
+
+Default: ``None``
+
+.. versionadded:: VERSION
+
+The maximum number of connections that a single :ref:`download handler
+<topics-download-handlers>` may keep open, whether they are being used for a
+request or kept around for reuse by a later request. Use ``0`` for no limit.
+
+``None`` lets Scrapy pick a limit that fits the system it runs on. How it is
+picked may change in any future Scrapy version, without a deprecation period,
+so set a number if you need it to stay the same.
+
+Scrapy currently picks half of the maximum number of open files allowed for the
+process, or 480 on Windows, where :func:`select.select` handles at most 512
+sockets.
+
+Each download handler applies this limit on its own, so a crawl of both
+``http://`` and ``https://`` URLs may open up to twice as many connections.
+:class:`~scrapy.core.downloader.handlers._httpx.HttpxDownloadHandler`
+additionally applies it once per proxy, since httpx `cannot use a different
+proxy per request <https://github.com/pydantic/httpx2/issues/818>`_.
+
+HTTP/2 multiplexes requests over a single connection per remote, so
+:class:`~scrapy.core.downloader.handlers.http2.H2DownloadHandler` reaches this
+limit only after as many remotes, regardless of concurrency, and it keeps a
+connection that is serving requests even if that leaves it over the limit.
+
+How the limit is reached also depends on the handler: some close the connection
+that has been idle the longest to make room for a new one, others make requests
+wait for a connection to free up. Either way, a value below
+:setting:`CONCURRENT_REQUESTS` keeps the crawl from reaching the configured
+concurrency, and logs a warning.
+
 .. setting:: CONCURRENT_ITEMS
 
 CONCURRENT_ITEMS
@@ -603,6 +641,21 @@ See also: :ref:`topics-autothrottle` and its
 It is possible to change this setting per domain by using
 :setting:`DOWNLOAD_SLOTS`.
 
+.. setting:: CONNECTION_KEEPALIVE_TIMEOUT
+
+CONNECTION_KEEPALIVE_TIMEOUT
+----------------------------
+
+Default: ``240.0``
+
+.. versionadded:: VERSION
+
+How long, in seconds, a connection is kept open for reuse after the response it
+was serving is complete. Use ``0`` to close connections as soon as they become
+idle.
+
+Servers usually close idle connections earlier than this on their own.
+
 .. setting:: DEFAULT_DROPITEM_LOG_LEVEL
 
 DEFAULT_DROPITEM_LOG_LEVEL
diff --git a/pyproject.toml b/pyproject.toml
index 0bdcf6b51..a403cd172 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -65,7 +65,7 @@ Tracker = "https://github.com/scrapy/scrapy/issues"
 [project.optional-dependencies]
 bpython = ["bpython>=0.7.1"]
 gcs = ["google-cloud-storage>=1.29.0"]
-httpx = ["httpx2[http2,socks]>=2.0.0"]
+httpx = ["httpx2[http2,socks]>=2.6.0"]
 images = ["Pillow>=8.3.2"]
 ipython = ["ipython>=8.15.0"]
 ptpython = ["ptpython>=3.0.23"]
diff --git a/scrapy/core/_http2/agent.py b/scrapy/core/_http2/agent.py
index 38be86b22..ce601c52b 100644
--- a/scrapy/core/_http2/agent.py
+++ b/scrapy/core/_http2/agent.py
@@ -6,12 +6,7 @@ from typing import TYPE_CHECKING
 from twisted.internet import defer
 from twisted.internet.defer import Deferred
 from twisted.python.failure import Failure
-from twisted.web.client import (
-    URI,
-    BrowserLikePolicyForHTTPS,
-    ResponseFailed,
-    _StandardEndpointFactory,
-)
+from twisted.web.client import URI, BrowserLikePolicyForHTTPS, _StandardEndpointFactory
 from twisted.web.error import SchemeNotSupported
 
 from scrapy.core._http2.protocol import H2ClientFactory, H2ClientProtocol
@@ -30,9 +25,10 @@ ConnectionKeyT = tuple[bytes, bytes, int]
 
 
 class H2ConnectionPool:
-    def __init__(self, reactor: ReactorBase, crawler: Crawler) -> None:
+    def __init__(self, reactor: ReactorBase, crawler: Crawler, limit: int = 0) -> None:
         self._reactor = reactor
         self._crawler = crawler
+        self._limit = limit
 
         # Store a dictionary which is used to get the respective
         # H2ClientProtocolInstance using the  key as Tuple(scheme, hostname, port)
@@ -52,15 +48,15 @@ class H2ConnectionPool:
     ) -> Deferred[H2ClientProtocol]:
         if key in self._pending_requests:
             # Received a request while connecting to remote
-            # Create a deferred which will fire with the H2ClientProtocol
-            # instance
-            d: Deferred[H2ClientProtocol] = Deferred()
-            self._pending_requests[key].append(d)
-            return d
+            return self._pending_request(key)
 
-        # Check if we already have a connection to the remote
-        conn = self._connections.get(key, None)
-        if conn:
+        # Check if we already have a usable connection to the remote, moving
+        # it to the end so that the pool stays ordered from least to most
+        # recently used
+        conn = self._connections.get(key)
+        if conn and not conn.closing:
+            del self._connections[key]
+            self._connections[key] = conn
             # Return this connection instance wrapped inside a deferred
             return defer.succeed(conn)
 
@@ -70,10 +66,10 @@ class H2ConnectionPool:
     def _new_connection(
         self, key: ConnectionKeyT, uri: URI, endpoint: HostnameEndpoint
     ) -> Deferred[H2ClientProtocol]:
+        self._enforce_limit()
         self._pending_requests[key] = deque()
 
         conn_lost_deferred: Deferred[list[BaseException]] = Deferred()
-        conn_lost_deferred.addCallback(self._remove_connection, key)
 
         factory = H2ClientFactory(
             uri,
@@ -82,16 +78,56 @@ class H2ConnectionPool:
             tls_verbose_logging=self._tls_verbose_logging,
         )
         conn_d = endpoint.connect(factory)
-        conn_d.addCallback(self.put_connection, key)
-
-        d: Deferred[H2ClientProtocol] = Deferred()
-        self._pending_requests[key].append(d)
+        d = self._pending_request(key)
+        conn_d.addCallbacks(
+            self.put_connection,
+            self._connection_failed,
+            callbackArgs=(key, conn_lost_deferred),
+            errbackArgs=(key,),
+        )
         return d
 
+    def _pending_request(self, key: ConnectionKeyT) -> Deferred[H2ClientProtocol]:
+        """Return a deferred that fires with the connection being established
+        for *key*.
+
+        Cancelling it takes it out of the queue, so that the connection, once
+        established or found unreachable, only fires the deferreds of the
+        requests that are still waiting for it.
+        """
+        pending_requests = self._pending_requests[key]
+        d: Deferred[H2ClientProtocol] = Deferred(pending_requests.remove)
+        pending_requests.append(d)
+        return d
+
+    def _enforce_limit(self) -> None:
+        """Close the least recently used connections that are not serving any
+        stream, to make room for one more connection.
+
+        HTTP/2 multiplexes requests over a single connection per remote, so a
+        connection with active streams is kept even if that leaves the pool
+        over the limit.
+        """
+        if not self._limit:
+            return
+        surplus = len(self._connections) + len(self._pending_requests) + 1 - self._limit
+        for key, conn in list(self._connections.items()):
+            if surplus <= 0:
+                return
+            if conn.metadata["active_streams"]:
+                continue
+            del self._connections[key]
+            conn.close_idle()
+            surplus -= 1
+
     def put_connection(
-        self, conn: H2ClientProtocol, key: ConnectionKeyT
+        self,
+        conn: H2ClientProtocol,
+        key: ConnectionKeyT,
+        conn_lost_deferred: Deferred[list[BaseException]],
     ) -> H2ClientProtocol:
         self._connections[key] = conn
+        conn_lost_deferred.addCallback(self._remove_connection, key, conn)
 
         # Now as we have established a proper HTTP/2 connection
         # we fire all the deferred's with the connection instance
@@ -102,16 +138,20 @@ class H2ConnectionPool:
 
         return conn
 
-    def _remove_connection(
-        self, errors: list[BaseException], key: ConnectionKeyT
-    ) -> None:
-        self._connections.pop(key)
-
-        # Call the errback of all the pending requests for this connection
-        pending_requests = self._pending_requests.pop(key, None)
+    def _connection_failed(self, failure: Failure, key: ConnectionKeyT) -> None:
+        """Fail the requests waiting for a connection that could not be
+        established, and let a later request try to connect again."""
+        pending_requests = self._pending_requests.pop(key)
         while pending_requests:
             d = pending_requests.popleft()
-            d.errback(ResponseFailed(errors))
+            d.errback(failure)
+
+    def _remove_connection(
+        self, errors: list[BaseException], key: ConnectionKeyT, conn: H2ClientProtocol
+    ) -> None:
+        # a newer connection may have taken over the key already
+        if self._connections.get(key) is conn:
+            del self._connections[key]
 
     def close_connections(self) -> None:
         """Close all the HTTP/2 connections and remove them from pool."""
diff --git a/scrapy/core/_http2/protocol.py b/scrapy/core/_http2/protocol.py
index 1e673ce8b..9470c94cf 100644
--- a/scrapy/core/_http2/protocol.py
+++ b/scrapy/core/_http2/protocol.py
@@ -86,8 +86,6 @@ class MethodNotAllowed405(H2Error):
 
 @implementer(IHandshakeListener)
 class H2ClientProtocol(Protocol, TimeoutMixin):
-    IDLE_TIMEOUT = 240
-
     def __init__(
         self,
         uri: URI,
@@ -109,6 +107,10 @@ class H2ClientProtocol(Protocol, TimeoutMixin):
         self._crawler: Crawler = crawler
         self._conn_lost_deferred: Deferred[list[BaseException]] = conn_lost_deferred
         self._tls_verbose_logging: bool = tls_verbose_logging
+        self._idle_timeout: float = crawler.settings.getfloat(
+            "CONNECTION_KEEPALIVE_TIMEOUT"
+        )
+        self.closing: bool = False
 
         config = H2Configuration(client_side=True, header_encoding="utf-8")
         self.conn = H2Connection(config=config)
@@ -195,6 +197,8 @@ class H2ClientProtocol(Protocol, TimeoutMixin):
         stream = self.streams.pop(stream_id)
         self.metadata["active_streams"] -= 1
         self._send_pending_requests()
+        if not self._idle_timeout and not self.metadata["active_streams"]:
+            self.close_idle()
         return stream
 
     def _new_stream(self, request: Request, spider: Spider) -> Stream:
@@ -254,7 +258,7 @@ class H2ClientProtocol(Protocol, TimeoutMixin):
         sending some data now: we should open with the connection preamble.
         """
         # Initialize the timeout
-        self.setTimeout(self.IDLE_TIMEOUT)  # type: ignore[no-untyped-call]
+        self.setTimeout(self._idle_timeout or None)  # type: ignore[no-untyped-call]
 
         assert self.transport is not None  # typing
         destination = self.transport.getPeer()
@@ -270,10 +274,17 @@ class H2ClientProtocol(Protocol, TimeoutMixin):
     def _lose_connection_with_error(self, errors: list[BaseException]) -> None:
         """Helper function to lose the connection with the error sent as a
         reason"""
+        self.closing = True
         self._conn_lost_errors.extend(errors)
         assert self.transport is not None  # typing
         self.transport.loseConnection()
 
+    def close_idle(self) -> None:
+        """Close a connection that is not serving any stream."""
+        self.conn.close_connection(error_code=ErrorCodes.NO_ERROR)
+        self._write_to_transport()
+        self._lose_connection_with_error([])
+
     def handshakeCompleted(self) -> None:
         """
         Close the connection if it's not made via the expected protocol
@@ -348,7 +359,7 @@ class H2ClientProtocol(Protocol, TimeoutMixin):
         self._lose_connection_with_error(
             [
                 DownloadTimeoutError(
-                    f"Connection was IDLE for more than {self.IDLE_TIMEOUT}s"
+                    f"Connection was IDLE for more than {self._idle_timeout}s"
                 )
             ]
         )
diff --git a/scrapy/core/downloader/handlers/_base_http.py b/scrapy/core/downloader/handlers/_base_http.py
index 83d130462..930c22285 100644
--- a/scrapy/core/downloader/handlers/_base_http.py
+++ b/scrapy/core/downloader/handlers/_base_http.py
@@ -1,12 +1,55 @@
 from __future__ import annotations
 
+import logging
+import sys
 from abc import ABC
 from typing import TYPE_CHECKING
 
 from .base import BaseDownloadHandler
 
+try:
+    # stdlib's resource module is only available on unix platforms
+    import resource
+except ImportError:
+    resource = None  # type: ignore[assignment]
+
 if TYPE_CHECKING:
     from scrapy.crawler import Crawler
+    from scrapy.settings import BaseSettings
+
+
+logger = logging.getLogger(__name__)
+
+# The number of sockets that select() accepts in CPython's Windows build; see
+# FD_SETSIZE in Modules/selectmodule.c.
+_FD_SETSIZE = 512
+
+# Room left for the sockets that a Windows crawl registers with the event loop
+# outside of download handlers, e.g. the asyncio self-pipe and the telnet
+# console.
+_WINDOWS_RESERVED_SOCKETS = 32
+
+
+def _auto_connection_limit() -> int:
+    if sys.platform == "win32":
+        return _FD_SETSIZE - _WINDOWS_RESERVED_SOCKETS
+    soft_limit: int = resource.getrlimit(resource.RLIMIT_NOFILE)[0]
+    if soft_limit == resource.RLIM_INFINITY:
+        return 0
+    return soft_limit // 2
+
+
+def _get_connection_limit(settings: BaseSettings) -> int:
+    limit = settings.get("CONCURRENT_CONNECTIONS_PER_HANDLER")
+    limit = _auto_connection_limit() if limit is None else int(limit)
+    concurrent_requests = settings.getint("CONCURRENT_REQUESTS")
+    if 0 < limit < concurrent_requests:
+        logger.warning(
+            f"CONCURRENT_CONNECTIONS_PER_HANDLER ({limit}) is lower than"
+            f" CONCURRENT_REQUESTS ({concurrent_requests}), which keeps the"
+            f" crawl from reaching the configured concurrency."
+        )
+    return limit
 
 
 class BaseHttpDownloadHandler(BaseDownloadHandler, ABC):
@@ -23,3 +66,11 @@ class BaseHttpDownloadHandler(BaseDownloadHandler, ABC):
             "DOWNLOADER_CLIENT_TLS_VERBOSE_LOGGING"
         )
         self._fail_on_dataloss_warned: bool = False
+        # these are useful for many handlers but used in different ways by them
+        self._pool_size_total: int = _get_connection_limit(crawler.settings)
+        self._pool_size_per_host: int = crawler.settings.getint(
+            "CONCURRENT_REQUESTS_PER_DOMAIN"
+        )
+        self._keepalive_timeout: float = crawler.settings.getfloat(
+            "CONNECTION_KEEPALIVE_TIMEOUT"
+        )
diff --git a/scrapy/core/downloader/handlers/_base_streaming.py b/scrapy/core/downloader/handlers/_base_streaming.py
index 5a669c732..d580baf1d 100644
--- a/scrapy/core/downloader/handlers/_base_streaming.py
+++ b/scrapy/core/downloader/handlers/_base_streaming.py
@@ -84,11 +84,6 @@ class BaseStreamingDownloadHandler(BaseHttpDownloadHandler, ABC, Generic[_Respon
             crawler.settings.get("DOWNLOAD_BIND_ADDRESS")
         )
         self._proxy_auth_encoding: str = crawler.settings.get("HTTPPROXY_AUTH_ENCODING")
-        # these are useful for many handlers but used in different ways by them
-        self._pool_size_total: int = crawler.settings.getint("CONCURRENT_REQUESTS")
-        self._pool_size_per_host: int = crawler.settings.getint(
-            "CONCURRENT_REQUESTS_PER_DOMAIN"
-        )
 
     @staticmethod
     @abstractmethod
diff --git a/scrapy/core/downloader/handlers/_httpx.py b/scrapy/core/downloader/handlers/_httpx.py
index 8bbffb233..fa712d119 100644
--- a/scrapy/core/downloader/handlers/_httpx.py
+++ b/scrapy/core/downloader/handlers/_httpx.py
@@ -93,10 +93,15 @@ class HttpxDownloadHandler(_Base):
         self._bind_host: str | None = self._get_bind_address_host()
         self._limits: httpx.Limits = httpx.Limits(
             # hard limit on simultaneous connections (None for no limit, which
-            # is what a CONCURRENT_REQUESTS of 0 means)
+            # is what a CONCURRENT_CONNECTIONS_PER_HANDLER of 0 means)
             max_connections=self._pool_size_total or None,
-            # total number of idle connections in the pool (extra ones are closed)
-            max_keepalive_connections=self._pool_size_total or None,
+            # total number of idle connections in the pool (extra ones are
+            # closed); keeping none of them is how keepalive gets disabled,
+            # since expiry alone is subject to the resolution of the clock
+            max_keepalive_connections=(
+                (self._pool_size_total or None) if self._keepalive_timeout else 0
+            ),
+            keepalive_expiry=self._keepalive_timeout,
         )
 
         self._default_client: httpx.AsyncClient = self._make_client()
diff --git a/scrapy/core/downloader/handlers/http11.py b/scrapy/core/downloader/handlers/http11.py
index c288ae792..285c47c81 100644
--- a/scrapy/core/downloader/handlers/http11.py
+++ b/scrapy/core/downloader/handlers/http11.py
@@ -66,6 +66,8 @@ from scrapy.utils.url import add_http_if_no_scheme
 from ._base_http import BaseHttpDownloadHandler
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from twisted.internet.base import ReactorBase
     from twisted.internet.interfaces import IAddress, IConsumer
     from twisted.web._newclient import Request as TxRequest
@@ -90,6 +92,160 @@ class _ResultT(TypedDict):
     stop_download: NotRequired[StopDownload | None]
 
 
+class _LenientHTTPClientParser(HTTPClientParser):
+    """Response parser that skips bad response header lines, those with no
+    colon in them, instead of failing to parse the whole response.
+
+    Some servers send such lines, and web browsers skip them and keep parsing
+    the header lines that follow. See
+    https://github.com/scrapy/scrapy/issues/210.
+    """
+
+    def lineReceived(self, line: bytes) -> None:
+        # A copy of twisted.web._newclient.HTTPParser.lineReceived() where the
+        # header name and value are only extracted from header lines that have
+        # a colon.
+
+        # Handle the normal CR LF case.
+        if line[-1:] == b"\r":
+            line = line[:-1]
+
+        if self.state == STATUS:
+            self.statusReceived(line)  # type: ignore[no-untyped-call]
+            self.state = HEADER
+            return
+
+        # HEADER is the only other state in which lines are received, as the
+        # parser switches to raw mode for the response body.
+        if not line or line[0] not in b" \t":
+            if self._partialHeader is not None:
+                header = b"".join(self._partialHeader)
+                if b":" in header:
+                    name, value = header.split(b":", 1)
+                    self.headerReceived(name, value.strip())  # type: ignore[no-untyped-call]
+                else:
+                    logger.debug(
+                        f"Skipping the bad response header line {header!r}, as "
+                        f"it has no colon."
+                    )
+            if not line:
+                # Empty line means the header section is over.
+                self.allHeadersReceived()  # type: ignore[no-untyped-call]
+            else:
+                # Line not beginning with LWS is another header.
+                self._partialHeader = [line]
+        else:
+            # A line beginning with LWS is a continuation of a header begun on
+            # a previous line.
+            self._partialHeader.append(line)  # type: ignore[union-attr]
+
+
+class _LenientHTTP11ClientProtocol(HTTP11ClientProtocol):
+    """Protocol that parses responses with :class:`_LenientHTTPClientParser`."""
+
+    def request(self, request: TxRequest) -> Deferred[IResponse]:
+        d: Deferred[IResponse] = super().request(request)
+        # HTTP11ClientProtocol.request() hardcodes the parser class, so the
+        # only way to use a different one is to replace the class of the parser
+        # object that it creates. This is safe because
+        # _LenientHTTPClientParser defines no additional state. The parser is
+        # always there because HTTPConnectionPool only reuses connections whose
+        # protocol is in the QUIESCENT state, for which request() always
+        # creates a parser.
+        assert self._parser is not None
+        self._parser.__class__ = _LenientHTTPClientParser
+        return d
+
+
+class _TrackedHTTP11ClientProtocol(_LenientHTTP11ClientProtocol):
+    """Connection that tells the pool that opened it when it closes, whether it
+    was cached at the time or in use."""
+
+    _pool: _BoundedHTTPConnectionPool
+    _pool_key: Any
+
+    def connectionLost(self, reason: Failure = connectionDone) -> None:
+        self._pool._forget(self)
+        super().connectionLost(reason)  # type: ignore[no-untyped-call]
+
+
+class _BoundedHTTP11ClientFactory(_HTTP11ClientFactory):
+    noisy = False
+
+    def __init__(
+        self,
+        quiescentCallback: Callable[[HTTP11ClientProtocol], None],
+        metadata: str,
+        pool: _BoundedHTTPConnectionPool,
+        key: Any,
+    ):
+        super().__init__(quiescentCallback, metadata)  # type: ignore[no-untyped-call]
+        self._pool: _BoundedHTTPConnectionPool = pool
+        self._key: Any = key
+
+    def buildProtocol(self, addr: IAddress | None) -> _TrackedHTTP11ClientProtocol:
+        protocol = _TrackedHTTP11ClientProtocol(self._quiescentCallback)  # type: ignore[no-untyped-call]
+        protocol._pool = self._pool
+        protocol._pool_key = self._key
+        self._pool._open.add(protocol)
+        return protocol
+
+
+class _BoundedHTTPConnectionPool(HTTPConnectionPool):
+    """Connection pool that keeps the number of open connections, cached and in
+    use alike, under *limit*, where 0 means no limit.
+
+    Opening a connection once the limit is reached closes the least recently
+    used cached connection to make room for it. When there is no cached
+    connection left to close, the limit is exceeded rather than requests
+    delayed.
+    """
+
+    _factory = _BoundedHTTP11ClientFactory  # type: ignore[assignment]
+
+    def __init__(self, reactor: ReactorBase, limit: int, keepalive_timeout: float):
+        super().__init__(reactor, persistent=bool(keepalive_timeout))  # type: ignore[no-untyped-call]
+        self.cachedConnectionTimeout = keepalive_timeout  # type: ignore[assignment]
+        self._limit: int = limit
+        self._open: set[_TrackedHTTP11ClientProtocol] = set()
+
+    def _newConnection(
+        self, key: Any, endpoint: TCP4ClientEndpoint
+    ) -> Deferred[HTTP11ClientProtocol]:
+        while self._limit and len(self._open) >= self._limit and self._evict():
+            pass
+
+        def quiescentCallback(protocol: HTTP11ClientProtocol) -> None:
+            self._putConnection(key, protocol)  # type: ignore[no-untyped-call]
+
+        factory = self._factory(quiescentCallback, repr(endpoint), self, key)
+        return endpoint.connect(factory)
+
+    def _forget(self, connection: _TrackedHTTP11ClientProtocol) -> None:
+        self._open.discard(connection)
+
+    def _evict(self) -> bool:
+        """Close the least recently used cached connection, and report whether
+        there was one to close.
+
+        ``_timeouts`` holds cached connections only, and ``getConnection()``
+        drops the ones it hands out, so its insertion order goes from least to
+        most recently used.
+        """
+        connection = next(iter(self._timeouts), None)
+        if connection is None:
+            return False
+        self._timeouts[connection].cancel()
+        # dropped early so that a caller looping on this method sees the room
+        # made by connections that have yet to finish closing
+        self._forget(connection)
+        key = connection._pool_key
+        self._removeConnection(key, connection)  # type: ignore[no-untyped-call]
+        if not self._connections[key]:
+            del self._connections[key]
+        return True
+
+
 class HTTP11DownloadHandler(BaseHttpDownloadHandler):
     def __init__(self, crawler: Crawler):
         if not crawler.settings.getbool("TWISTED_REACTOR_ENABLED"):
@@ -99,11 +255,10 @@ class HTTP11DownloadHandler(BaseHttpDownloadHandler):
 
         from twisted.internet import reactor
 
-        self._pool: HTTPConnectionPool = HTTPConnectionPool(reactor, persistent=True)
-        self._pool.maxPersistentPerHost = crawler.settings.getint(
-            "CONCURRENT_REQUESTS_PER_DOMAIN"
+        self._pool: _BoundedHTTPConnectionPool = _BoundedHTTPConnectionPool(
+            reactor, self._pool_size_total, self._keepalive_timeout
         )
-        self._pool._factory = _LenientHTTP11ClientFactory
+        self._pool.maxPersistentPerHost = self._pool_size_per_host
 
         self._contextFactory: IPolicyForHTTPS = _load_context_factory_from_settings(
             crawler
@@ -145,7 +300,7 @@ class HTTP11DownloadHandler(BaseHttpDownloadHandler):
     async def close(self) -> None:
         from twisted.internet import reactor
 
-        d: Deferred[None] = self._pool.closeCachedConnections()
+        d: Deferred[None] = self._pool.closeCachedConnections()  # type: ignore[no-untyped-call]
         # closeCachedConnections will hang on network or server issues, so
         # we'll manually timeout the deferred.
         #
@@ -748,77 +903,3 @@ class _ResponseReader(Protocol):
             reason = Failure(exc)
 
         self._finished.errback(reason)
-
-
-class _LenientHTTPClientParser(HTTPClientParser):
-    """Response parser that skips bad response header lines, those with no
-    colon in them, instead of failing to parse the whole response.
-
-    Some servers send such lines, and web browsers skip them and keep parsing
-    the header lines that follow. See
-    https://github.com/scrapy/scrapy/issues/210.
-    """
-
-    def lineReceived(self, line: bytes) -> None:
-        # A copy of twisted.web._newclient.HTTPParser.lineReceived() where the
-        # header name and value are only extracted from header lines that have
-        # a colon.
-
-        # Handle the normal CR LF case.
-        if line[-1:] == b"\r":
-            line = line[:-1]
-
-        if self.state == STATUS:
-            self.statusReceived(line)  # type: ignore[no-untyped-call]
-            self.state = HEADER
-            return
-
-        # HEADER is the only other state in which lines are received, as the
-        # parser switches to raw mode for the response body.
-        if not line or line[0] not in b" \t":
-            if self._partialHeader is not None:
-                header = b"".join(self._partialHeader)
-                if b":" in header:
-                    name, value = header.split(b":", 1)
-                    self.headerReceived(name, value.strip())  # type: ignore[no-untyped-call]
-                else:
-                    logger.debug(
-                        f"Skipping the bad response header line {header!r}, as "
-                        f"it has no colon."
-                    )
-            if not line:
-                # Empty line means the header section is over.
-                self.allHeadersReceived()  # type: ignore[no-untyped-call]
-            else:
-                # Line not beginning with LWS is another header.
-                self._partialHeader = [line]
-        else:
-            # A line beginning with LWS is a continuation of a header begun on
-            # a previous line.
-            self._partialHeader.append(line)  # type: ignore[union-attr]
-
-
-class _LenientHTTP11ClientProtocol(HTTP11ClientProtocol):
-    """Protocol that parses responses with :class:`_LenientHTTPClientParser`."""
-
-    def request(self, request: TxRequest) -> Deferred[IResponse]:
-        d: Deferred[IResponse] = super().request(request)
-        # HTTP11ClientProtocol.request() hardcodes the parser class, so the
-        # only way to use a different one is to replace the class of the parser
-        # object that it creates. This is safe because
-        # _LenientHTTPClientParser defines no additional state. The parser is
-        # always there because HTTPConnectionPool only reuses connections whose
-        # protocol is in the QUIESCENT state, for which request() always
-        # creates a parser.
-        assert self._parser is not None
-        self._parser.__class__ = _LenientHTTPClientParser
-        return d
-
-
-class _LenientHTTP11ClientFactory(_HTTP11ClientFactory):
-    """Factory that builds :class:`_LenientHTTP11ClientProtocol` protocols."""
-
-    noisy = False
-
-    def buildProtocol(self, addr: IAddress | None) -> HTTP11ClientProtocol:
-        return _LenientHTTP11ClientProtocol(self._quiescentCallback)  # type: ignore[no-untyped-call]
diff --git a/scrapy/core/downloader/handlers/http2.py b/scrapy/core/downloader/handlers/http2.py
index 935a2ae4c..e159d2dac 100644
--- a/scrapy/core/downloader/handlers/http2.py
+++ b/scrapy/core/downloader/handlers/http2.py
@@ -6,7 +6,7 @@ from urllib.parse import urldefrag
 
 from scrapy.core._http2.agent import H2Agent, H2ConnectionPool
 from scrapy.core.downloader.contextfactory import _load_context_factory_from_settings
-from scrapy.core.downloader.handlers.base import BaseDownloadHandler
+from scrapy.core.downloader.handlers._base_http import BaseHttpDownloadHandler
 from scrapy.exceptions import (
     DownloadTimeoutError,
     NotConfigured,
@@ -29,7 +29,7 @@ if TYPE_CHECKING:
     from scrapy.spiders import Spider
 
 
-class H2DownloadHandler(BaseDownloadHandler):
+class H2DownloadHandler(BaseHttpDownloadHandler):
     lazy = True
 
     def __init__(self, crawler: Crawler):
@@ -40,7 +40,7 @@ class H2DownloadHandler(BaseDownloadHandler):
 
         from twisted.internet import reactor
 
-        self._pool = H2ConnectionPool(reactor, crawler)
+        self._pool = H2ConnectionPool(reactor, crawler, self._pool_size_total)
         self._context_factory = _load_context_factory_from_settings(crawler)
         self._bind_address = crawler.settings.get("DOWNLOAD_BIND_ADDRESS")
 
diff --git a/scrapy/settings/default_settings.py b/scrapy/settings/default_settings.py
index b97844103..45bdcd93e 100644
--- a/scrapy/settings/default_settings.py
+++ b/scrapy/settings/default_settings.py
@@ -43,9 +43,11 @@ __all__ = [
     "CLOSESPIDER_TIMEOUT_NO_ITEM",
     "COMMANDS_MODULE",
     "COMPRESSION_ENABLED",
+    "CONCURRENT_CONNECTIONS_PER_HANDLER",
     "CONCURRENT_ITEMS",
     "CONCURRENT_REQUESTS",
     "CONCURRENT_REQUESTS_PER_DOMAIN",
+    "CONNECTION_KEEPALIVE_TIMEOUT",
     "COOKIES_DEBUG",
     "COOKIES_ENABLED",
     "CRAWLSPIDER_FOLLOW_LINKS",
@@ -250,11 +252,15 @@ COMMANDS_MODULE = ""
 
 COMPRESSION_ENABLED = True
 
+CONCURRENT_CONNECTIONS_PER_HANDLER = None
+
 CONCURRENT_ITEMS = 100
 
 CONCURRENT_REQUESTS = 16
 CONCURRENT_REQUESTS_PER_DOMAIN = 8
 
+CONNECTION_KEEPALIVE_TIMEOUT = 240.0
+
 COOKIES_ENABLED = True
 COOKIES_DEBUG = False
 
diff --git a/tests/mockserver/http.py b/tests/mockserver/http.py
index 6074f1475..350153393 100644
--- a/tests/mockserver/http.py
+++ b/tests/mockserver/http.py
@@ -18,6 +18,7 @@ from .http_resources import (
     ChunkedResource,
     ClientIPResource,
     Compress,
+    ConnectionId,
     ContentLengthHeaderResource,
     Delay,
     Drop,
@@ -60,6 +61,7 @@ class Root(BaseResource):
         put_child(self, b"static", File(str(Path(tests_datadir, "test_site/"))))
         put_child(self, b"redirect-to", RedirectTo())
         put_child(self, b"text", Data(b"Works", "text/plain"))
+        put_child(self, b"connection-id", ConnectionId())
         put_child(
             self,
             b"html",
diff --git a/tests/mockserver/http_resources.py b/tests/mockserver/http_resources.py
index 969e29a8c..68c14af83 100644
--- a/tests/mockserver/http_resources.py
+++ b/tests/mockserver/http_resources.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import gzip
+import itertools
 import json
 import random
 from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar
@@ -450,3 +451,35 @@ class SetCookie(BaseResource):
                 cookie = (cookie_name.decode() + "=" + cookie_value.decode()).encode()
                 request.setHeader(b"Set-Cookie", cookie)
         return b""
+
+
+_connection_ids = itertools.count()
+
+
+class ConnectionId(LeafResource):
+    """Return the identifier of the connection serving the request, after the
+    number of seconds given in the ``delay`` argument.
+
+    Two responses carrying the same identifier were served over the same
+    connection, which lets tests tell connection reuse from reconnection.
+    """
+
+    def render_GET(self, request: Request) -> int | bytes:
+        delay = getarg(request, b"delay", 0, type_=float)
+        if not delay:
+            return self._connection_id(request)
+        self.deferRequest(request, delay, self._delayedRender, request)
+        return NOT_DONE_YET
+
+    def _delayedRender(self, request: Request) -> None:
+        request.write(self._connection_id(request))
+        request.finish()
+
+    @staticmethod
+    def _connection_id(request: Request) -> bytes:
+        # under HTTP/2 the channel of a request is its stream, so the
+        # connection has to be reached through it
+        channel: Any = getattr(request.channel, "_conn", request.channel)
+        if not hasattr(channel, "_mockserver_connection_id"):
+            channel._mockserver_connection_id = next(_connection_ids)
+        return to_bytes(str(channel._mockserver_connection_id))
diff --git a/tests/test_downloader_handler_httpx.py b/tests/test_downloader_handler_httpx.py
index 27a44227d..21326a4b0 100644
--- a/tests/test_downloader_handler_httpx.py
+++ b/tests/test_downloader_handler_httpx.py
@@ -167,13 +167,23 @@ class TestRealWebsite(HttpxDownloadHandlerMixin, TestRealWebsiteBase):
     pass
 
 
-@pytest.mark.parametrize(("concurrency", "expected"), [(16, 16), (0, None)])
+@pytest.mark.parametrize(("limit", "expected"), [(16, 16), (0, None)])
 @coroutine_test
-async def test_pool_limits(concurrency: int, expected: int | None) -> None:
-    crawler = get_crawler(settings_dict={"CONCURRENT_REQUESTS": concurrency})
+async def test_pool_limits(limit: int, expected: int | None) -> None:
+    crawler = get_crawler(settings_dict={"CONCURRENT_CONNECTIONS_PER_HANDLER": limit})
     handler = build_from_crawler(HttpxDownloadHandler, crawler)
     try:
         assert handler._limits.max_connections == expected
         assert handler._limits.max_keepalive_connections == expected
     finally:
         await handler.close()
+
+
+@coroutine_test
+async def test_keepalive_timeout() -> None:
+    crawler = get_crawler(settings_dict={"CONNECTION_KEEPALIVE_TIMEOUT": 5})
+    handler = build_from_crawler(HttpxDownloadHandler, crawler)
+    try:
+        assert handler._limits.keepalive_expiry == 5
+    finally:
+        await handler.close()
diff --git a/tests/test_downloader_handler_twisted_http11.py b/tests/test_downloader_handler_twisted_http11.py
index db353750d..d930faf15 100644
--- a/tests/test_downloader_handler_twisted_http11.py
+++ b/tests/test_downloader_handler_twisted_http11.py
@@ -3,15 +3,27 @@
 from __future__ import annotations
 
 import sys
+from contextlib import asynccontextmanager
 from typing import TYPE_CHECKING, Any
 
 import pytest
+from twisted.internet.defer import DeferredList
+
+try:
+    import resource
+except ImportError:
+    resource = None  # type: ignore[assignment]
 
 from scrapy import Spider
+from scrapy.core.downloader.handlers._base_http import _auto_connection_limit
 from scrapy.core.downloader.handlers.http11 import HTTP11DownloadHandler
 from scrapy.crawler import Crawler
 from scrapy.exceptions import NotConfigured
+from scrapy.http import Request
+from scrapy.utils.defer import deferred_from_coro, maybe_deferred_to_future
 from scrapy.utils.misc import build_from_crawler
+from scrapy.utils.spider import DefaultSpider
+from scrapy.utils.test import get_crawler
 from tests.utils.bases.download_handlers_http import (
     TestHttpBase,
     TestHttpProxyBase,
@@ -26,9 +38,13 @@ from tests.utils.bases.download_handlers_http import (
     TestRealWebsiteBase,
     TestSimpleHttpsBase,
 )
+from tests.utils.decorators import coroutine_test
 
 if TYPE_CHECKING:
+    from collections.abc import AsyncGenerator
+
     from scrapy.core.downloader.handlers import DownloadHandlerProtocol
+    from tests.mockserver.http import MockServer
 
 
 pytestmark = pytest.mark.requires_reactor  # HTTP11DownloadHandler requires a reactor
@@ -55,6 +71,89 @@ def test_not_configured_without_reactor() -> None:
         build_from_crawler(HTTP11DownloadHandler, crawler)
 
 
+@asynccontextmanager
+async def _get_dh(
+    settings_dict: dict[str, Any],
+) -> AsyncGenerator[HTTP11DownloadHandler]:
+    crawler = get_crawler(DefaultSpider, settings_dict)
+    crawler.spider = crawler._create_spider()
+    dh = build_from_crawler(HTTP11DownloadHandler, crawler)
+    try:
+        yield dh
+    finally:
+        await dh.close()
+
+
+@coroutine_test
+async def test_connection_limit_auto() -> None:
+    async with _get_dh({}) as dh:
+        assert dh._pool._limit == _auto_connection_limit()
+
+
+@coroutine_test
+@pytest.mark.parametrize("limit", [0, 20])
+async def test_connection_limit_explicit(limit: int) -> None:
+    async with _get_dh({"CONCURRENT_CONNECTIONS_PER_HANDLER": limit}) as dh:
+        assert dh._pool._limit == limit
+
+
+@coroutine_test
+async def test_connection_limit_below_concurrent_requests(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    settings_dict = {
+        "CONCURRENT_CONNECTIONS_PER_HANDLER": 4,
+        "CONCURRENT_REQUESTS": 8,
+    }
+    with caplog.at_level("WARNING"):
+        async with _get_dh(settings_dict):
+            pass
+    assert "CONCURRENT_CONNECTIONS_PER_HANDLER (4)" in caplog.text
+
+
+@pytest.mark.skipif(resource is None, reason="No file descriptor limit")
+@coroutine_test
+async def test_connection_limit_auto_without_file_descriptor_limit(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    infinity = resource.RLIM_INFINITY
+    monkeypatch.setattr(resource, "getrlimit", lambda _: (infinity, infinity))
+    async with _get_dh({}) as dh:
+        assert dh._pool._limit == 0
+
+
+@coroutine_test
+async def test_connection_limit_with_several_connections_per_host(
+    mockserver: MockServer,
+) -> None:
+    url = mockserver.url("/connection-id")
+    slow_url = mockserver.url("/connection-id?delay=0.5")
+    other_url = url.replace("127.0.0.1", "localhost")
+    async with _get_dh({"CONCURRENT_CONNECTIONS_PER_HANDLER": 2}) as dh:
+        results = await maybe_deferred_to_future(
+            DeferredList(
+                [
+                    deferred_from_coro(dh.download_request(Request(slow_url)))
+                    for _ in range(2)
+                ],
+                fireOnOneErrback=True,
+            )
+        )
+        ids = {response.text for _, response in results}
+        assert len(ids) == 2
+        # only one of the two connections to the host makes room for the
+        # connection to the other host, so the other one stays reusable
+        await dh.download_request(Request(other_url))
+        reused = await dh.download_request(Request(url))
+    assert reused.text in ids
+
+
+@coroutine_test
+async def test_keepalive_timeout() -> None:
+    async with _get_dh({"CONNECTION_KEEPALIVE_TIMEOUT": 5}) as dh:
+        assert dh._pool.cachedConnectionTimeout == 5
+
+
 class TestHttp(HTTP11DownloadHandlerMixin, TestHttpBase):
     pass
 
diff --git a/tests/test_downloader_handler_twisted_http2.py b/tests/test_downloader_handler_twisted_http2.py
index 95e1ff633..258ede04e 100644
--- a/tests/test_downloader_handler_twisted_http2.py
+++ b/tests/test_downloader_handler_twisted_http2.py
@@ -7,12 +7,14 @@ import sys
 from typing import TYPE_CHECKING, Any
 
 import pytest
+from twisted.internet.defer import DeferredList
 from twisted.web.http import H2_ENABLED
 
 from scrapy import Spider
 from scrapy.crawler import Crawler
 from scrapy.exceptions import DownloadFailedError, NotConfigured
 from scrapy.http import Request
+from scrapy.utils.defer import deferred_from_coro, maybe_deferred_to_future
 from scrapy.utils.misc import build_from_crawler
 from tests.utils.bases.download_handlers_http import (
     TestHttpProxyBase,
@@ -81,22 +83,6 @@ class TestHttp2(H2DownloadHandlerMixin, TestHttpsBase):
             response = await download_handler.download_request(request)
         assert response.protocol == "h2"
 
-    def test_download_conn_failed(self) -> None:  # type: ignore[override]
-        # Unlike HTTP11DownloadHandler which raises it from download_request()
-        # (without any special handling), here ConnectionRefusedError (raised in
-        # twisted.internet.endpoints.startConnectionAttempts()) bubbles up as
-        # an unhandled exception in a Deferred and the handler waits until
-        # DOWNLOAD_TIMEOUT.
-        pytest.skip("The handler doesn't properly reraise ConnectionRefusedError")
-
-    def test_download_dns_error(self) -> None:  # type: ignore[override]
-        # Unlike HTTP11DownloadHandler which raises it from download_request()
-        # (without any special handling), here DNSLookupError (raised in
-        # twisted.internet.endpoints.startConnectionAttempts()) bubbles up as
-        # an unhandled exception in a Deferred and the handler waits until
-        # DOWNLOAD_TIMEOUT.
-        pytest.skip("The handler doesn't properly reraise DNSLookupError")
-
     @coroutine_test
     async def test_concurrent_requests_same_domain(
         self, mockserver: MockServer
@@ -111,6 +97,23 @@ class TestHttp2(H2DownloadHandlerMixin, TestHttpsBase):
             response2 = await download_handler.download_request(request2)
             assert response2.headers["Content-Length"] == b"79"
 
+    @coroutine_test
+    async def test_parallel_requests_same_domain(self, mockserver: MockServer) -> None:
+        url = mockserver.url("/connection-id", is_secure=self.is_secure)
+        async with self.get_dh() as download_handler:
+            results = await maybe_deferred_to_future(
+                DeferredList(
+                    [
+                        deferred_from_coro(download_handler.download_request(request))
+                        for request in (Request(url), Request(url))
+                    ],
+                    fireOnOneErrback=True,
+                )
+            )
+        # the second request waited for the connection that the first one was
+        # opening instead of opening a second one
+        assert len({response.text for _, response in results}) == 1
+
     @pytest.mark.xfail(reason="https://github.com/python-hyper/h2/issues/1247")
     @coroutine_test
     async def test_connect_request(self, mockserver: MockServer) -> None:
diff --git a/tests/test_http2_client_protocol.py b/tests/test_http2_client_protocol.py
index e96a5484e..bfd8754d5 100644
--- a/tests/test_http2_client_protocol.py
+++ b/tests/test_http2_client_protocol.py
@@ -756,11 +756,9 @@ class TestHttps2ClientProtocol:
             yield make_request_dfd(client, request)
 
         for err in exc_info.value.reasons:
-            from scrapy.core._http2.protocol import H2ClientProtocol  # noqa: PLC0415
-
             if isinstance(err, DownloadTimeoutError):
                 assert (
-                    f"Connection was IDLE for more than {H2ClientProtocol.IDLE_TIMEOUT}s"
+                    f"Connection was IDLE for more than {client._idle_timeout}s"
                     in str(err)
                 )
                 break
diff --git a/tests/test_settings/__init__.py b/tests/test_settings/__init__.py
index 97c095ae0..88622ddac 100644
--- a/tests/test_settings/__init__.py
+++ b/tests/test_settings/__init__.py
@@ -758,6 +758,15 @@ def test_remove_from_list(before, name, item, after):
     assert settings.getpriority(name) == expected_settings.getpriority(name)
 
 
+def test_default_settings_all():
+    deprecated = {"DNS_RESOLVER"}
+    assert scrapy_default_settings.__all__ == sorted(
+        name
+        for name in vars(scrapy_default_settings)
+        if name.isupper() and name not in deprecated
+    )
+
+
 def test_deprecated_dns_resolver_setting():
     settings = Settings()
     with pytest.warns(
diff --git a/tests/utils/bases/download_handlers_http.py b/tests/utils/bases/download_handlers_http.py
index abf153817..a01f74658 100644
--- a/tests/utils/bases/download_handlers_http.py
+++ b/tests/utils/bases/download_handlers_http.py
@@ -16,6 +16,7 @@ from urllib.parse import urlparse
 
 import pytest
 from cryptography.x509 import load_der_x509_certificate
+from twisted.internet.defer import DeferredList
 from twisted.internet.ssl import Certificate
 from twisted.python.failure import Failure
 
@@ -108,6 +109,69 @@ class TestHttpBase(ABC):
         finally:
             await dh.close()
 
+    async def _connection_ids(
+        self, urls: list[str], settings_dict: dict[str, Any]
+    ) -> set[str]:
+        """Request *urls* one after another and return the identifiers of the
+        connections that the server used to serve them."""
+        async with self.get_dh(settings_dict) as dh:
+            responses = [await dh.download_request(Request(url)) for url in urls]
+        return {response.text for response in responses}
+
+    @staticmethod
+    def _other_host(url: str) -> str:
+        """Return *url* pointing at the same server through a different name,
+        so that its connection cannot be reused."""
+        return url.replace("127.0.0.1", "localhost")
+
+    @coroutine_test
+    async def test_keepalive(self, mockserver: MockServer) -> None:
+        url = mockserver.url("/connection-id", is_secure=self.is_secure)
+        assert len(await self._connection_ids([url, url], {})) == 1
+
+    @coroutine_test
+    async def test_no_keepalive(self, mockserver: MockServer) -> None:
+        url = mockserver.url("/connection-id", is_secure=self.is_secure)
+        settings_dict = {"CONNECTION_KEEPALIVE_TIMEOUT": 0}
+        assert len(await self._connection_ids([url, url], settings_dict)) == 2
+
+    @coroutine_test
+    async def test_connection_limit(self, mockserver: MockServer) -> None:
+        url = mockserver.url("/connection-id", is_secure=self.is_secure)
+        urls = [url, self._other_host(url)] * 2
+
+        assert len(await self._connection_ids(urls, {})) == 2
+
+        settings_dict = {"CONCURRENT_CONNECTIONS_PER_HANDLER": 1}
+        assert len(await self._connection_ids(urls, settings_dict)) == 4
+
+        settings_dict = {"CONCURRENT_CONNECTIONS_PER_HANDLER": 0}
+        assert len(await self._connection_ids(urls, settings_dict)) == 2
+
+    @coroutine_test
+    async def test_connection_limit_spares_connections_in_use(
+        self, mockserver: MockServer
+    ) -> None:
+        url = mockserver.url("/connection-id", is_secure=self.is_secure)
+        slow_url = mockserver.url("/connection-id?delay=0.5", is_secure=self.is_secure)
+        other_url = self._other_host(url)
+        async with self.get_dh({"CONCURRENT_CONNECTIONS_PER_HANDLER": 1}) as dh:
+            first = await dh.download_request(Request(url))
+            results = await maybe_deferred_to_future(
+                DeferredList(
+                    [
+                        deferred_from_coro(dh.download_request(Request(slow_url))),
+                        deferred_from_coro(dh.download_request(Request(other_url))),
+                    ],
+                    fireOnOneErrback=True,
+                )
+            )
+        ids = {response.text for _, response in results}
+        # the connection serving the slow request was kept even though the
+        # request to the other host had to exceed the limit to get one
+        assert first.text in ids
+        assert len(ids) == 2
+
     @coroutine_test
     async def test_unsupported_scheme(self) -> None:
         request = Request("unsupp://unsupported.scheme")
diff --git a/tox.ini b/tox.ini
index 7ea1c57d7..34af7bac3 100644
--- a/tox.ini
+++ b/tox.ini
@@ -142,7 +142,7 @@ deps =
     brotlicffi==1.2.0.0; implementation_name == "pypy"
     cryptography==37.0.0
     cssselect==0.9.1
-    httpx2==2.0.0
+    httpx2==2.6.0
     itemadapter==0.1.0
     lxml==4.6.4
     parsel==1.5.0
@@ -191,7 +191,7 @@ deps =
     boto3==1.20.0
     bpython==0.7.1
     google-cloud-storage==1.29.0
-    httpx2[http2,socks]==2.0.0
+    httpx2[http2,socks]==2.6.0
     ipython==8.15.0
     ptpython==3.0.23
     robotexclusionrulesparser==1.6.2