Raise StreamClosedError for evicted streams

src/h2/connection.py

@@ -793,14 +793,17 @@ def send_headers(self,
         )
 
         # Check we can open the stream.
-        if stream_id not in self.streams:
+        if stream_id not in self.streams and stream_id not in self._closed_streams:
             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:
                 msg = f"Max outbound streams is {max_open_streams}, {value} open"
                 raise TooManyStreamsError(msg)
 
         self.state_machine.process_input(ConnectionInputs.SEND_HEADERS)
+        if stream_id in self._closed_streams:
+            raise StreamClosedError(stream_id)
+
         stream = self._get_or_create_stream(
             stream_id, AllowedStreamIDs(self.config.client_side),
         )

tests/test_closed_streams.py

@@ -105,6 +105,37 @@ def test_closed_stream_not_present_in_streams_dict(self, frame_factory) -> None:
         # The streams dictionary should be empty.
         assert not c.streams
 
+    def test_sending_headers_on_evicted_reset_stream(self, frame_factory) -> None:
+        """
+        Sending headers on a reset stream raises StreamClosedError even after
+        the stream has been evicted from the active streams dictionary.
+        """
+        c = h2.connection.H2Connection(config=self.server_config)
+        c.receive_data(frame_factory.preamble())
+        c.initiate_connection()
+
+        f = frame_factory.build_headers_frame(self.example_request_headers)
+        c.receive_data(f.serialize())
+
+        f = frame_factory.build_rst_stream_frame(stream_id=1)
+        c.receive_data(f.serialize())
+
+        # Force the closed stream into the bounded closed-stream cache.
+        assert not c.open_inbound_streams
+        assert 1 not in c.streams
+
+        with pytest.raises(h2.exceptions.StreamClosedError) as exc_info:
+            c.send_headers(1, self.example_response_headers)
+
+        assert exc_info.value.stream_id == 1
+
+        c.close_connection()
+        with pytest.raises(
+            h2.exceptions.ProtocolError,
+            match="Invalid input ConnectionInputs.SEND_HEADERS",
+        ):
+            c.send_headers(1, self.example_response_headers)
+
     def test_receive_rst_stream_on_closed_stream(self, frame_factory) -> None:
         """
         RST_STREAM frame should be ignored if stream is in a closed state.