Drain open streams after receiving a graceful GOAWAY

CHANGELOG.rst

@@ -1,6 +1,28 @@
 Release History
 ===============
 
+4.4.2 (unreleased)
+------------------
+
+**API Changes (Backward Compatible)**
+
+- Support graceful shutdown by the remote peer (RFC 9113, Section 6.8). After
+  a GOAWAY with error code ``NO_ERROR``, the connection drains: its state is
+  ``CLOSED`` and no new local streams may be opened, but frames for open
+  streams are still processed, PING and SETTINGS frames are still
+  acknowledged, and the outbound data buffer is kept. The client and server
+  rules of the open connection still apply. A further graceful GOAWAY keeps
+  draining; a GOAWAY with an error code, an invalid input or
+  ``close_connection()`` ends the drain, and a connection that is already
+  closed is never reopened by a later GOAWAY. ``H2Connection.state_machine.draining``
+  reports the drain. ``ConnectionTerminated`` is still fired for every GOAWAY.
+  (#1181, #1199)
+
+**Bugfixes**
+
+- Ignore leftover frames received after the connection has closed, instead of
+  raising ``ProtocolError``. Further GOAWAY frames are still processed.
+
 4.4.1 (2026-08-03)
 ------------------
 

src/h2/connection.py

@@ -247,8 +247,58 @@ class H2ConnectionStateMachine:
             (None, ConnectionState.CLOSED),
     }
 
+    # Inputs that stay valid in the CLOSED state while the connection drains
+    # after the remote peer sent a graceful GOAWAY (error code NO_ERROR).
+    # RFC 9113, Section 6.8: streams that were already open may still
+    # complete, but the receiver of the GOAWAY must not open new streams.
+    # SEND_PUSH_PROMISE is therefore absent; H2Connection refuses other new
+    # local streams, since it knows the stream IDs. An input is accepted only
+    # if the open connection would also accept it (see _draining_state).
+    _draining_inputs = frozenset({
+        ConnectionInputs.SEND_HEADERS,
+        ConnectionInputs.SEND_DATA,
+        ConnectionInputs.SEND_WINDOW_UPDATE,
+        ConnectionInputs.SEND_PING,
+        ConnectionInputs.SEND_SETTINGS,
+        ConnectionInputs.SEND_RST_STREAM,
+        ConnectionInputs.SEND_PRIORITY,
+        ConnectionInputs.SEND_ALTERNATIVE_SERVICE,
+        ConnectionInputs.RECV_HEADERS,
+        ConnectionInputs.RECV_PUSH_PROMISE,
+        ConnectionInputs.RECV_DATA,
+        ConnectionInputs.RECV_WINDOW_UPDATE,
+        ConnectionInputs.RECV_PING,
+        ConnectionInputs.RECV_SETTINGS,
+        ConnectionInputs.RECV_RST_STREAM,
+        ConnectionInputs.RECV_PRIORITY,
+        ConnectionInputs.RECV_ALTERNATIVE_SERVICE,
+    })
+
     def __init__(self) -> None:
         self.state = ConnectionState.IDLE
+        # While draining: the state the connection would be in had the GOAWAY
+        # not arrived (IDLE, CLIENT_OPEN or SERVER_OPEN). It keeps the normal
+        # client and server rules in force. None when not draining.
+        self._draining_state: ConnectionState | None = None
+
+    @property
+    def draining(self) -> bool:
+        """
+        Whether the connection is draining: the remote peer sent a graceful
+        GOAWAY, so the state is CLOSED and this side may not open new streams,
+        but frames for open streams are still processed. The peer that sent
+        the GOAWAY may still open streams (RFC 9113, Section 6.8).
+        """
+        return self._draining_state is not None
+
+    def _start_draining(self, open_state: ConnectionState) -> None:
+        """
+        Start (or continue) draining after a graceful GOAWAY from the peer.
+        ``open_state`` is the state before the GOAWAY. It must not be CLOSED:
+        a connection that is already closed stays closed.
+        """
+        assert open_state is not ConnectionState.CLOSED
+        self._draining_state = open_state
 
     def process_input(self, input_: ConnectionInputs) -> list[Event]:
         """
@@ -258,6 +308,16 @@ def process_input(self, input_: ConnectionInputs) -> list[Event]:
             msg = "Input must be an instance of ConnectionInputs"
             raise ValueError(msg)  # noqa: TRY004
 
+        if self._draining_state is not None:
+            transition = self._transitions.get((self._draining_state, input_))
+            if input_ in self._draining_inputs and transition is not None:
+                self._draining_state = transition[1]
+                return []
+            # Anything else (a GOAWAY, or an input the open connection would
+            # also reject) ends the drain. A further graceful GOAWAY from the
+            # peer continues it (see H2Connection._receive_goaway_frame).
+            self._draining_state = None
+
         try:
             func, target_state = self._transitions[(self.state, input_)]
         except KeyError as e:
@@ -573,6 +633,8 @@ def initiate_upgrade_connection(self, settings_header: bytes | None = None) -> b
             "Upgrade connection. Current settings: %s", self.local_settings,
         )
 
+        self._refuse_new_stream_while_draining(1)
+
         frame_data = None
         # Begin by getting the preamble in place.
         self.initiate_connection()
@@ -609,6 +671,19 @@ def initiate_upgrade_connection(self, settings_header: bytes | None = None) -> b
         self.streams[1].upgrade(self.config.client_side)
         return frame_data
 
+    def _refuse_new_stream_while_draining(self, stream_id: int) -> None:
+        """
+        The receiver of a GOAWAY must not open new streams (RFC 9113, Section
+        6.8). Raise before any state changes, so the drain continues for the
+        streams that are already open.
+        """
+        if self.state_machine.draining:
+            msg = (
+                f"Cannot open stream {stream_id}: the remote peer sent GOAWAY, "
+                "so no new streams may be opened"
+            )
+            raise ProtocolError(msg)
+
     def _get_or_create_stream(self, stream_id: int, allowed_ids: AllowedStreamIDs) -> H2Stream:
         """
         Gets a stream by its stream ID. Will create one if one does not already
@@ -794,6 +869,7 @@ def send_headers(self,
 
         # Check we can open the stream.
         if stream_id not in self.streams:
+            self._refuse_new_stream_while_draining(stream_id)
             max_open_streams = self.remote_settings.max_concurrent_streams
             value = self.open_outbound_streams # take a copy due to the property accessor having side affects
             if (value + 1) > max_open_streams:
@@ -1010,6 +1086,7 @@ def push_stream(self,
             msg = "Remote peer has disabled stream push"
             raise ProtocolError(msg)
 
+        self._refuse_new_stream_while_draining(promised_stream_id)
         self.state_machine.process_input(ConnectionInputs.SEND_PUSH_PROMISE)
         stream = self._get_stream_by_id(stream_id)
 
@@ -1541,6 +1618,15 @@ def _receive_frame(self, frame: Frame) -> list[Event]:
         """
         events: list[Event]
         self.config.logger.trace("Received frame: %s", repr(frame))
+        # Leftover frames can still be in the TCP buffer after we closed.
+        # Extra GOAWAYs are already valid; anything else is not a protocol error.
+        # While draining after a graceful GOAWAY, frames are still processed.
+        if (
+            self.state_machine.state == ConnectionState.CLOSED
+            and not self.state_machine.draining
+            and not isinstance(frame, GoAwayFrame)
+        ):
+            return []
         try:
             # I don't love using __class__ here, maybe reconsider it.
             frames, events = self._frame_dispatch_table[frame.__class__](frame)
@@ -1894,12 +1980,28 @@ def _receive_goaway_frame(self, frame: GoAwayFrame) -> tuple[list[Frame], list[E
         """
         Receive a GOAWAY frame on the connection.
         """
+        # The state the connection would be in without this GOAWAY: the state
+        # before it, or the state an existing drain keeps. None if the
+        # connection is already closed for good (an error GOAWAY, a protocol
+        # error, or close_connection()); a later graceful GOAWAY must not
+        # revive it (RFC 9113, Section 5.4.1).
+        open_state: ConnectionState | None = self.state_machine._draining_state
+        if open_state is None and self.state_machine.state is not ConnectionState.CLOSED:
+            open_state = self.state_machine.state
+
         events = self.state_machine.process_input(
             ConnectionInputs.RECV_GOAWAY,
         )
 
-        # Clear the outbound data buffer: we cannot send further data now.
-        self.clear_outbound_data_buffer()
+        if frame.error_code == ErrorCodes.NO_ERROR and open_state is not None:
+            # A graceful shutdown (RFC 9113, Section 6.8): no new streams, but
+            # streams that are already open may complete. Keep the outbound
+            # buffer, which may hold WINDOW_UPDATE, SETTINGS or PING ACK frames
+            # those streams need.
+            self.state_machine._start_draining(open_state)
+        else:
+            # Clear the outbound data buffer: we cannot send further data now.
+            self.clear_outbound_data_buffer()
 
         # Fire an appropriate ConnectionTerminated event.
         new_event = ConnectionTerminated()

src/h2/events.py

@@ -562,8 +562,22 @@ def __repr__(self) -> str:
 class ConnectionTerminated(Event):
     """
     The ConnectionTerminated event is fired when a connection is torn down by
-    the remote peer using a GOAWAY frame. Once received, no further action may
-    be taken on the connection: a new connection must be established.
+    the remote peer using a GOAWAY frame. Once received, no new streams may be
+    opened on the connection: a new connection must be established for them.
+
+    If ``error_code`` is ``NO_ERROR``, this is a graceful shutdown (RFC 9113,
+    Section 6.8): streams up to ``last_stream_id`` may still complete. On a
+    connection that was not already closed, h2 starts (or continues) a drain
+    and keeps processing frames for open streams; check
+    ``H2Connection.state_machine.draining`` for the current state. Locally initiated
+    streams above ``last_stream_id`` were not processed by the peer and are
+    safe to retry elsewhere. h2 does not reset or filter them, and does not
+    check that a later GOAWAY keeps ``last_stream_id`` from increasing: the
+    application decides. With any other error code, no further action may be
+    taken on the connection.
+
+    .. versionchanged:: 4.4.2
+       A graceful GOAWAY no longer prevents open streams from completing.
     """
 
     def __init__(self) -> None:

tests/test_basic_logic.py

@@ -1788,6 +1788,35 @@ def test_receiving_goaway_frame(self, frame_factory) -> None:
 
         assert not c.data_to_send()
 
+    def test_frames_after_goaway_are_ignored(self, frame_factory) -> None:
+        """
+        Frames already in the receive buffer after a GOAWAY with an error code
+        must not raise ProtocolError. Further GOAWAY frames are still accepted.
+        (After a graceful GOAWAY the connection drains instead; see
+        test_graceful_goaway.py.)
+        """
+        c = h2.connection.H2Connection(config=self.server_config)
+        c.initiate_connection()
+        c.receive_data(frame_factory.preamble())
+        c.clear_outbound_data_buffer()
+
+        goaway = frame_factory.build_goaway_frame(
+            last_stream_id=0, error_code=h2.errors.ErrorCodes.PROTOCOL_ERROR,
+        )
+        events = c.receive_data(goaway.serialize())
+        assert len(events) == 1
+        assert isinstance(events[0], h2.events.ConnectionTerminated)
+
+        ping = frame_factory.build_ping_frame(b"\x01\x02\x03\x04\x05\x06\x07\x08")
+        events = c.receive_data(ping.serialize())
+        assert events == []
+        assert not c.data_to_send()
+        assert c.state_machine.state == h2.connection.ConnectionState.CLOSED
+
+        events = c.receive_data(goaway.serialize())
+        assert len(events) == 1
+        assert isinstance(events[0], h2.events.ConnectionTerminated)
+
     def test_receiving_multiple_goaway_frames(self, frame_factory) -> None:
         """
         Multiple GOAWAY frames can be received at once, and are allowed. Each