http2: stream the request body to the application

.github/workflows/docker-integration.yml

@@ -6,11 +6,19 @@ on:
     paths:
       - 'gunicorn/uwsgi/**'
       - 'tests/docker/uwsgi/**'
+      - 'gunicorn/http2/**'
+      - 'gunicorn/asgi/**'
+      - 'gunicorn/workers/**'
+      - 'tests/docker/h2spec/**'
       - '.github/workflows/docker-integration.yml'
   pull_request:
     paths:
       - 'gunicorn/uwsgi/**'
       - 'tests/docker/uwsgi/**'
+      - 'gunicorn/http2/**'
+      - 'gunicorn/asgi/**'
+      - 'gunicorn/workers/**'
+      - 'tests/docker/h2spec/**'
       - '.github/workflows/docker-integration.yml'
 
 permissions:
@@ -43,3 +51,21 @@ jobs:
       - name: Run uWSGI integration tests
         run: |
           pytest tests/docker/uwsgi/ -v --tb=short
+
+  h2spec:
+    name: HTTP/2 conformance (h2spec)
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    steps:
+      - uses: actions/checkout@v7
+      - name: Set up Python
+        uses: actions/setup-python@v7
+        with:
+          python-version: "3.12"
+      - name: Install test dependencies
+        run: |
+          python -m pip install --upgrade pip
+          python -m pip install pytest pytest-cov
+      - name: Run h2spec against each HTTP/2 worker
+        run: |
+          pytest tests/docker/h2spec/ -v --tb=short

.gitignore

@@ -19,3 +19,4 @@ setuptools-*
 site/
 docs/site/
 tests/docker/*/results/
+tests/docker/h2spec/certs/

docs/content/2026-news.md

@@ -1,6 +1,41 @@
 <span id="news-2026"></span>
 # Changelog - 2026
 
+## Unreleased
+
+### Bug Fixes
+
+- **HTTP/2 request bodies were buffered in full before dispatch**: DATA
+  frames were kept in two buffers, copied twice more when the request was
+  built, and flow control credit went back to the peer as each frame arrived,
+  so a client that never sent END_STREAM could grow a worker without the
+  application being called. A request is now dispatched on its headers and
+  `wsgi.input` (or ASGI `receive()`) pulls the body from the stream as it
+  arrives, returning window credit only for what the application has read.
+  A peer can have no more than the receive window in flight per stream. A
+  body the application leaves unread is cut off with `RST_STREAM(NO_ERROR)`
+  once the response is sent. On the ASGI worker, streams on a connection are
+  served concurrently. Only listeners with `h2` in `http_protocols` were
+  affected.
+- **HTTP/2 review fixes**: a review of the HTTP/2 support found these, all
+  fixed on every h2-capable worker unless noted. A graceful `GOAWAY` from the
+  peer is honoured wherever it lands in a read: established streams finish,
+  later ones are refused, and no private h2 state is touched. Waiting for
+  send credit is bounded by `timeout` instead of a fixed five seconds; a
+  peer that stops reading gets `RST_STREAM(CANCEL)` and the application is
+  stopped rather than told the response succeeded. An application error
+  after the response headers went out resets the stream instead of sending a
+  second `HEADERS` block, which corrupted the HPACK table for every later
+  response. An empty final body chunk now ends the stream. On `gthread` and
+  `gevent`, frames read while waiting for credit are handled in order, an
+  idle connection is closed after `keepalive` and a body that stalls for
+  `timeout` is cancelled, so a peer can no longer pin a thread, and streams
+  the peer resets before they are served are dropped so `HEADERS+RST_STREAM`
+  floods cannot grow the worker. On the ASGI worker, `receive()` after the
+  body blocks until the peer goes away instead of spinning the event loop, a
+  reader paused under backpressure is resumed, and in-flight streams get the
+  disconnect grace period before they are cancelled.
+
 ## 26.2.0 - 2026-08-24
 
 ### New Features

docs/content/guides/http2.md

@@ -775,6 +775,7 @@ protections against known vulnerabilities.
 |--------|------------|---------|
 | Stream Multiplexing Abuse | Limit concurrent streams | `http2_max_concurrent_streams` (default: 100) |
 | HPACK Bomb | Header size limits | `http2_max_header_list_size` (default: 65536) |
+| Unbounded Upload | Body streams to the app; window credit returns as it reads | `http2_initial_window_size` (default: 65535) |
 | Large Frame Attack | Frame size limits | `http2_max_frame_size` (validated: 16384-16777215) |
 | Resource Exhaustion | Flow control windows | `http2_initial_window_size` (default: 65535) |
 | Slow Read (Slowloris) | Connection timeouts | `timeout` and `keepalive` settings |

docs/content/reference/settings.md

@@ -1945,6 +1945,14 @@ For the non sync workers it just means that the worker process is still
 communicating and is not tied to the length of time required to handle a
 single request.
 
+On HTTP/2 connections it also bounds a single stream: a request body
+that makes no progress for this long, or a response the peer stops
+reading for this long, is cancelled with ``RST_STREAM``. 0 disables
+that too.
+
+!!! info "Changed in 26.3.0"
+    Also bounds a stalled HTTP/2 stream.
+
 ### `graceful_timeout`
 
 **Command line:** `--graceful-timeout INT`
@@ -1974,6 +1982,12 @@ set this to a higher value.
     ``sync`` worker does not support persistent connections and will
     ignore this option.
 
+On the ``gthread`` and ``gevent`` workers an HTTP/2 connection with no
+stream open is closed with ``GOAWAY`` after this many seconds.
+
+!!! info "Changed in 26.3.0"
+    Also applies to idle HTTP/2 connections.
+
 ### `asgi_loop`
 
 **Command line:** `--asgi-loop STRING`

gunicorn/asgi/protocol.py

@@ -22,6 +22,8 @@
 from gunicorn.asgi.uwsgi import AsyncUWSGIRequest
 from gunicorn.http.errors import NoMoreData
 from gunicorn.http2 import negotiation
+from gunicorn.http2.errors import HTTP2ErrorCode, HTTP2StreamError
+from gunicorn.http2.stream import StreamState
 from gunicorn.uwsgi.errors import UWSGIParseException
 
 
@@ -357,6 +359,8 @@ def __init__(self, worker):
         self.cfg = worker.cfg
         self.log = worker.log
         self.app = worker.asgi
+        # Tasks serving HTTP/2 streams on this connection
+        self._h2_tasks = set()
 
         self.transport = None
         self.reader = None  # Only used for HTTP/2
@@ -865,6 +869,11 @@ def connection_lost(self, exc):
         if self._body_receiver is not None:
             self._body_receiver.signal_disconnect()
 
+        # HTTP/2: every stream task learns the peer is gone
+        h2_conn = getattr(self, '_h2_conn', None)
+        if h2_conn is not None:
+            h2_conn.abort_streams_nowait()
+
         # Schedule task cancellation after grace period if task doesn't complete
         if self._task and not self._task.done():
             grace_period = getattr(self.cfg, 'asgi_disconnect_grace_period', 3)
@@ -1682,12 +1691,20 @@ async def _handle_http2_connection(self, transport, ssl_object, upgrade=None):
                 upgraded = await h2_conn.initiate_upgrade(
                     settings, http1_req, body)
                 self._h2_conn = h2_conn
-                await self._serve_http2_request(
-                    upgraded, h2_conn, sockname, peername)
+                self._start_http2_stream(upgraded, h2_conn, sockname, peername)
                 self.worker.nr += 1
 
-            # Main loop - receive and handle requests
-            while not h2_conn.is_closed and self.worker.alive:
+            # Main loop: each request is served by its own task as soon
+            # as its headers are in, while this loop keeps reading frames
+            # so bodies stream to the tasks and window credit flows back.
+            # Once the worker stops accepting, or the peer sends a graceful
+            # GOAWAY, the loop keeps reading until the streams already in
+            # flight have finished, so their bodies still arrive; streams
+            # opened meanwhile are refused.
+            while not h2_conn.is_closed:
+                if (not self.worker.alive or h2_conn.draining) and not self._h2_tasks:
+                    break
+                self._maybe_resume_reading()
                 try:
                     requests = await h2_conn.receive_data(timeout=1.0)
                 except asyncio.TimeoutError:
@@ -1697,30 +1714,74 @@ async def _handle_http2_connection(self, transport, ssl_object, upgrade=None):
                     break
 
                 for req in requests:
-                    await self._serve_http2_request(
-                        req, h2_conn, sockname, peername)
-
-                # Increment worker request count
-                self.worker.nr += len(requests)
-
-                # Check max_requests
-                if self.worker.nr >= self.worker.max_requests:
-                    self.log.info("Autorestarting worker after current request.")
-                    self.worker.alive = False
-                    break
+                    stream_id = req.stream.stream_id
+                    if not self.worker.alive:
+                        await h2_conn.reset_stream(
+                            stream_id, HTTP2ErrorCode.REFUSED_STREAM)
+                        h2_conn.cleanup_stream(stream_id)
+                        continue
+                    self._start_http2_stream(req, h2_conn, sockname, peername)
+                    self.worker.nr += 1
+                    if self.worker.nr >= self.worker.max_requests:
+                        self.log.info("Autorestarting worker after current request.")
+                        self.worker.alive = False
 
         except asyncio.CancelledError:
             pass
         except Exception as e:
             self.log.exception("HTTP/2 connection error: %s", e)
         finally:
+            await self._finish_http2_streams()
             if hasattr(self, '_h2_conn'):
                 try:
                     await self._h2_conn.close()
                 except Exception:
                     pass
             self._close_transport()
 
+    async def _finish_http2_streams(self):
+        """Tell in-flight streams the peer is gone, then cancel stragglers.
+
+        Same contract as HTTP/1: the application sees http.disconnect
+        first and gets asgi_disconnect_grace_period to wind down.
+        """
+        tasks = list(self._h2_tasks)
+        if not tasks:
+            return
+        h2_conn = getattr(self, '_h2_conn', None)
+        if h2_conn is not None:
+            h2_conn.abort_streams_nowait()
+        grace = getattr(self.cfg, 'asgi_disconnect_grace_period', 3)
+        if grace > 0:
+            await asyncio.wait(tasks, timeout=grace)
+        await self._cancel_http2_streams()
+
+    def _maybe_resume_reading(self):
+        """Resume a reader paused by data_received once the loop caught up.
+
+        Only the HTTP/1 loop used to resume; an HTTP/2 connection that
+        once paused never read again. Half the limit gives hysteresis.
+        """
+        if (self._reading_paused and self.reader is not None
+                and len(self.reader._buffer) <= self._max_buffer_size // 2):
+            self._resume_reading()
+
+    def _start_http2_stream(self, req, h2_conn, sockname, peername):
+        """Serve one stream in its own task."""
+        task = self.worker.loop.create_task(
+            self._serve_http2_request(req, h2_conn, sockname, peername))
+        self._h2_tasks.add(task)
+        task.add_done_callback(self._h2_tasks.discard)
+
+    async def _cancel_http2_streams(self):
+        """Stop stream tasks still running when the connection ends."""
+        tasks = list(self._h2_tasks)
+        if not tasks:
+            return
+        for task in tasks:
+            task.cancel()
+        await asyncio.gather(*tasks, return_exceptions=True)
+
     async def _serve_http2_request(self, req, h2_conn, sockname, peername):
         """Run one HTTP/2 request, answering 500 rather than dropping it."""
         try:
@@ -1769,11 +1830,24 @@ async def _handle_http2_request(self, request, h2_conn, sockname, peername):
         # Track if we've finished receiving body
         body_received = False
 
-        async def receive():
+        async def receive():  # pylint: disable=too-many-return-statements
             nonlocal body_received
 
-            # Check if stream is closed or missing
-            if stream is None or stream.state.name == "CLOSED":
+            # The peer reset the stream or the connection is gone
+            if stream is None or stream.disconnected:
+                return {"type": "http.disconnect"}
+            if not body_received and stream.state is StreamState.CLOSED:
+                return {"type": "http.disconnect"}
+
+            if body_received:
+                # The body has been delivered. Per the ASGI spec the next
+                # message is http.disconnect: at once if the response is
+                # already out, otherwise when the peer actually goes away.
+                # Returning again at once would spin the disconnect
+                # listeners frameworks run alongside the handler.
+                if response_complete or peer_gone:
+                    return {"type": "http.disconnect"}
+                await stream.wait_disconnect()
                 return {"type": "http.disconnect"}
 
             # First call: if body already complete (small requests), return it
@@ -1792,7 +1866,7 @@ async def receive():
                     stream.read_body_chunk(),
                     timeout=30.0
                 )
-            except asyncio.TimeoutError:
+            except (asyncio.TimeoutError, HTTP2StreamError):
                 return {"type": "http.disconnect"}
 
             if chunk is None:
@@ -1803,19 +1877,28 @@ async def receive():
                     "more_body": False,
                 }
 
-            if stream._body_complete:
+            # END_STREAM may have landed with several frames still queued.
+            done = stream._body_complete and not stream._body_chunks
+            if done:
                 body_received = True
 
             return {
                 "type": "http.request",
                 "body": chunk,
-                "more_body": not stream._body_complete,
+                "more_body": not done,
             }
 
-        async def send(message):
+        # Set once a send loses to the peer's RST_STREAM: the response is
+        # over, later sends are inert, and none of it is an app failure.
+        peer_gone = False
+
+        async def send(message):  # pylint: disable=too-many-return-statements
             nonlocal response_started, response_complete, headers_sent
             nonlocal response_status, response_headers, response_sent, exc_to_raise
-            nonlocal omits_body, omits_body_warned
+            nonlocal omits_body, omits_body_warned, peer_gone
+
+            if peer_gone:
+                return
 
             msg_type = message["type"]
 
@@ -1868,24 +1951,24 @@ async def send(message):
                     response_hdrs.extend(headers)
 
                     # Send headers without end_stream since we have body
-                    stream = h2_conn.streams.get(stream_id)
-                    if stream is None:
-                        exc_to_raise = RuntimeError("Stream closed")
+                    if not await h2_conn.send_response_headers(
+                            stream_id, response_hdrs, end_stream=False):
+                        peer_gone = response_complete = True
                         return
-                    h2_conn.h2_conn.send_headers(stream_id, response_hdrs, end_stream=False)
-                    stream.send_headers(response_hdrs, end_stream=False)
-                    await h2_conn._send_pending_data()
                     headers_sent = True
 
                 # Stream body immediately
                 if body:
-                    await h2_conn.send_data(stream_id, body, end_stream=not more_body)
+                    if not await h2_conn.send_data(stream_id, body, end_stream=not more_body):
+                        peer_gone = response_complete = True
+                        return
                     response_sent += len(body)
 
                 if not more_body:
                     if not body:
                         # Empty final chunk - send end_stream
-                        await h2_conn.send_data(stream_id, b"", end_stream=True)
+                        if not await h2_conn.send_data(stream_id, b"", end_stream=True):
+                            peer_gone = True
                     response_complete = True
 
             elif msg_type == "http.response.trailers":
@@ -1894,7 +1977,8 @@ async def send(message):
                     return
                 trailer_headers = message.get("headers", [])
                 trailers = self._convert_h2_headers(trailer_headers)
-                await h2_conn.send_trailers(stream_id, trailers)
+                if not await h2_conn.send_trailers(stream_id, trailers):
+                    peer_gone = True
 
         # Only build environ for logging if access logging is enabled
         access_log_enabled = self.log.access_log_enabled
@@ -1918,11 +2002,8 @@ async def send(message):
                 headers = self._convert_h2_headers(response_headers)
                 response_hdrs = [(':status', str(response_status))]
                 response_hdrs.extend(headers)
-                stream = h2_conn.streams.get(stream_id)
-                if stream:
-                    h2_conn.h2_conn.send_headers(stream_id, response_hdrs, end_stream=True)
-                    stream.send_headers(response_hdrs, end_stream=True)
-                    await h2_conn._send_pending_data()
+                await h2_conn.send_response_headers(
+                    stream_id, response_hdrs, end_stream=True)
 
         except Exception:
             self.log.exception("Error in ASGI application")