Merge pull request #1211 from Kriechi/fix-starved-flow-window

Kriechi · web-flow · commit 7be1f2805a74 · 2020-02-08T12:23:27.000+01:00
auto-emit WINDOW_UPDATE frames on unprocessed DATA frames
diff --git a/.travis.yml b/.travis.yml
@@ -28,6 +28,9 @@ matrix:
     - python: "3.7"
       dist: xenial
       env: TOXENV=py37
+    - python: "3.8"
+      dist: xenial
+      env: TOXENV=py38
     - python: "pypy-5.3.1"
       dist: trusty
       env: TOXENV=pypy
diff --git a/HISTORY.rst b/HISTORY.rst
@@ -4,6 +4,19 @@ Release History
 3.1.1dev0
 ---------
 
+Bugfixes
+~~~~~~~~
+
+- Receiving DATA frames on closed (or reset) streams now properly emit a
+  WINDOW_UPDATE to keep the connection flow window topped up.
+
+API Changes (Backward-Incompatible)
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+- ``h2.config.logger`` now uses a `trace(...)` function, in addition
+  to `debug(...)`. If you defined a custom logger object, you need to handle
+  these new function calls.
+
 
 3.1.1 (2019-08-02)
 ------------------
diff --git a/h2/config.py b/h2/config.py
@@ -29,7 +29,7 @@ class DummyLogger(object):
     """
     An Logger object that does not actual logging, hence a DummyLogger.
 
-    For the class the log operation is merely a no-op.  The intent is to avoid
+    For the class the log operation is merely a no-op. The intent is to avoid
     conditionals being sprinkled throughout the hyper-h2 code for calls to
     logging functions when no logger is passed into the corresponding object.
     """
@@ -42,6 +42,12 @@ def debug(self, *vargs, **kwargs):
         """
         pass
 
+    def trace(self, *vargs, **kwargs):
+        """
+        No-op logging. Only level needed for now.
+        """
+        pass
+
 
 class H2Configuration(object):
     """
diff --git a/h2/connection.py b/h2/connection.py
@@ -1450,7 +1450,7 @@ def receive_data(self, data):
         :returns: A list of events that the remote peer triggered by sending
             this data.
         """
-        self.config.logger.debug(
+        self.config.logger.trace(
             "Process received data on connection. Received data: %r", data
         )
 
@@ -1634,6 +1634,35 @@ def _receive_push_promise_frame(self, frame):
 
         return frames, events + stream_events
 
+    def _handle_data_on_closed_stream(self, events, exc, frame):
+        # This stream is already closed - and yet we received a DATA frame.
+        # The received DATA frame counts towards the connection flow window.
+        # We need to manually to acknowledge the DATA frame to update the flow
+        # window of the connection. Otherwise the whole connection stalls due
+        # the inbound flow window being 0.
+        frames = []
+        conn_manager = self._inbound_flow_control_window_manager
+        conn_increment = conn_manager.process_bytes(
+            frame.flow_controlled_length
+        )
+        if conn_increment:
+            f = WindowUpdateFrame(0)
+            f.window_increment = conn_increment
+            frames.append(f)
+            self.config.logger.debug(
+                "Received DATA frame on closed stream %d - "
+                "auto-emitted a WINDOW_UPDATE by %d",
+                frame.stream_id, conn_increment
+            )
+        f = RstStreamFrame(exc.stream_id)
+        f.error_code = exc.error_code
+        frames.append(f)
+        self.config.logger.debug(
+            "Stream %d already CLOSED or cleaned up - "
+            "auto-emitted a RST_FRAME" % frame.stream_id
+        )
+        return frames, events + exc._events
+
     def _receive_data_frame(self, frame):
         """
         Receive a data frame on the connection.
@@ -1646,12 +1675,19 @@ def _receive_data_frame(self, frame):
         self._inbound_flow_control_window_manager.window_consumed(
             flow_controlled_length
         )
-        stream = self._get_stream_by_id(frame.stream_id)
-        frames, stream_events = stream.receive_data(
-            frame.data,
-            'END_STREAM' in frame.flags,
-            flow_controlled_length
-        )
+
+        try:
+            stream = self._get_stream_by_id(frame.stream_id)
+            frames, stream_events = stream.receive_data(
+                frame.data,
+                'END_STREAM' in frame.flags,
+                flow_controlled_length
+            )
+        except StreamClosedError as e:
+            # This stream is either marked as CLOSED or already gone from our
+            # internal state.
+            return self._handle_data_on_closed_stream(events, e, frame)
+
         return frames, events + stream_events
 
     def _receive_settings_frame(self, frame):
diff --git a/setup.cfg b/setup.cfg
@@ -1,2 +1,5 @@
+[tool:pytest]
+testpaths = test
+
 [wheel]
 universal = 1
diff --git a/setup.py b/setup.py
@@ -58,18 +58,19 @@
         'Programming Language :: Python :: 2',
         'Programming Language :: Python :: 2.7',
         'Programming Language :: Python :: 3',
-        'Programming Language :: Python :: 3.3',
         'Programming Language :: Python :: 3.4',
         'Programming Language :: Python :: 3.5',
         'Programming Language :: Python :: 3.6',
+        'Programming Language :: Python :: 3.7',
+        'Programming Language :: Python :: 3.8',
         'Programming Language :: Python :: Implementation :: CPython',
         'Programming Language :: Python :: Implementation :: PyPy',
     ],
     install_requires=[
         'hyperframe>=5.2.0, <6',
-        'hpack>=2.3,<4',
+        'hpack>=3.0,<4',
     ],
     extras_require={
-        ':python_version == "2.7" or python_version == "3.3"': ['enum34>=1.1.6, <2'],
+        ':python_version == "2.7"': ['enum34>=1.1.6, <2'],
     }
 )
diff --git a/test/test_closed_streams.py b/test/test_closed_streams.py
@@ -195,7 +195,6 @@ class TestStreamsClosedByEndStream(object):
     @pytest.mark.parametrize(
         "frame",
         [
-            lambda self, ff: ff.build_data_frame(b'hello'),
             lambda self, ff: ff.build_headers_frame(
                 self.example_request_headers, flags=['END_STREAM']),
             lambda self, ff: ff.build_headers_frame(
@@ -245,7 +244,6 @@ def test_frames_after_recv_end_will_error(self,
     @pytest.mark.parametrize(
         "frame",
         [
-            lambda self, ff: ff.build_data_frame(b'hello'),
             lambda self, ff: ff.build_headers_frame(
                 self.example_response_headers, flags=['END_STREAM']),
             lambda self, ff: ff.build_headers_frame(
@@ -344,15 +342,15 @@ class TestStreamsClosedByRstStream(object):
                 self.example_request_headers),
             lambda self, ff: ff.build_headers_frame(
                 self.example_request_headers, flags=['END_STREAM']),
-            lambda self, ff: ff.build_data_frame(b'hello'),
         ]
     )
     def test_resets_further_frames_after_recv_reset(self,
                                                     frame_factory,
                                                     frame):
         """
         A stream that is closed by receive RST_STREAM can receive further
-        frames: it simply sends RST_STREAM for it.
+        frames: it simply sends RST_STREAM for it, and additionally
+        WINDOW_UPDATE for DATA frames.
         """
         c = h2.connection.H2Connection(config=self.server_config)
         c.receive_data(frame_factory.preamble())
@@ -396,14 +394,66 @@ def test_resets_further_frames_after_recv_reset(self,
         assert not events
         assert c.data_to_send() == rst_frame.serialize() * 3
 
+    def test_resets_further_data_frames_after_recv_reset(self,
+                                                         frame_factory):
+        """
+        A stream that is closed by receive RST_STREAM can receive further
+        DATA frames: it simply sends WINDOW_UPDATE for the connection flow
+        window, and RST_STREAM for the stream.
+        """
+        c = h2.connection.H2Connection(config=self.server_config)
+        c.receive_data(frame_factory.preamble())
+        c.initiate_connection()
+
+        header_frame = frame_factory.build_headers_frame(
+            self.example_request_headers, flags=['END_STREAM']
+        )
+        c.receive_data(header_frame.serialize())
+
+        c.send_headers(
+            stream_id=1,
+            headers=self.example_response_headers,
+            end_stream=False
+        )
+
+        rst_frame = frame_factory.build_rst_stream_frame(
+            1, h2.errors.ErrorCodes.STREAM_CLOSED
+        )
+        c.receive_data(rst_frame.serialize())
+        c.clear_outbound_data_buffer()
+
+        f = frame_factory.build_data_frame(
+            data=b'some data'
+        )
+
+        events = c.receive_data(f.serialize())
+        assert not events
+
+        expected = frame_factory.build_rst_stream_frame(
+            stream_id=1,
+            error_code=h2.errors.ErrorCodes.STREAM_CLOSED,
+        ).serialize()
+        assert c.data_to_send() == expected
+
+        events = c.receive_data(f.serialize() * 3)
+        assert not events
+        assert c.data_to_send() == expected * 3
+
+        # Iterate over the streams to make sure it's gone, then confirm the
+        # behaviour is unchanged.
+        c.open_outbound_streams
+
+        events = c.receive_data(f.serialize() * 3)
+        assert not events
+        assert c.data_to_send() == expected * 3
+
     @pytest.mark.parametrize(
         "frame",
         [
             lambda self, ff: ff.build_headers_frame(
                 self.example_request_headers),
             lambda self, ff: ff.build_headers_frame(
                 self.example_request_headers, flags=['END_STREAM']),
-            lambda self, ff: ff.build_data_frame(b'hello'),
         ]
     )
     def test_resets_further_frames_after_send_reset(self,
@@ -455,3 +505,51 @@ def test_resets_further_frames_after_send_reset(self,
         events = c.receive_data(f.serialize() * 3)
         assert not events
         assert c.data_to_send() == rst_frame.serialize() * 3
+
+    def test_resets_further_data_frames_after_send_reset(self,
+                                                         frame_factory):
+        """
+        A stream that is closed by sent RST_STREAM can receive further
+        data frames: it simply sends WINDOW_UPDATE and RST_STREAM for it.
+        """
+        c = h2.connection.H2Connection(config=self.server_config)
+        c.receive_data(frame_factory.preamble())
+        c.initiate_connection()
+
+        header_frame = frame_factory.build_headers_frame(
+            self.example_request_headers, flags=['END_STREAM']
+        )
+        c.receive_data(header_frame.serialize())
+
+        c.send_headers(
+            stream_id=1,
+            headers=self.example_response_headers,
+            end_stream=False
+        )
+
+        c.reset_stream(1, h2.errors.ErrorCodes.INTERNAL_ERROR)
+
+        c.clear_outbound_data_buffer()
+
+        f = frame_factory.build_data_frame(
+            data=b'some data'
+        )
+        events = c.receive_data(f.serialize())
+        assert not events
+        expected = frame_factory.build_rst_stream_frame(
+            stream_id=1,
+            error_code=h2.errors.ErrorCodes.STREAM_CLOSED,
+        ).serialize()
+        assert c.data_to_send() == expected
+
+        events = c.receive_data(f.serialize() * 3)
+        assert not events
+        assert c.data_to_send() == expected * 3
+
+        # Iterate over the streams to make sure it's gone, then confirm the
+        # behaviour is unchanged.
+        c.open_outbound_streams
+
+        events = c.receive_data(f.serialize() * 3)
+        assert not events
+        assert c.data_to_send() == expected * 3
diff --git a/test/test_flow_control_window.py b/test/test_flow_control_window.py
@@ -638,6 +638,42 @@ def test_reject_local_overlarge_increase_stream_window(self):
         with pytest.raises(h2.exceptions.FlowControlError):
             c.increment_flow_control_window(increment=increment, stream_id=1)
 
+    def test_send_update_on_closed_streams(self, frame_factory):
+        c = h2.connection.H2Connection()
+        c.initiate_connection()
+        c.send_headers(1, self.example_request_headers)
+        c.reset_stream(1)
+
+        c.clear_outbound_data_buffer()
+        c.open_outbound_streams
+        c.open_inbound_streams
+
+        f = frame_factory.build_data_frame(b'some data'*1500)
+        events = c.receive_data(f.serialize()*3)
+        assert not events
+
+        expected = frame_factory.build_rst_stream_frame(
+            stream_id=1,
+            error_code=h2.errors.ErrorCodes.STREAM_CLOSED,
+        ).serialize() * 2 + frame_factory.build_window_update_frame(
+            stream_id=0,
+            increment=40500,
+        ).serialize() + frame_factory.build_rst_stream_frame(
+            stream_id=1,
+            error_code=h2.errors.ErrorCodes.STREAM_CLOSED,
+        ).serialize()
+        assert c.data_to_send() == expected
+
+        f = frame_factory.build_data_frame(b'')
+        events = c.receive_data(f.serialize())
+        assert not events
+
+        expected = frame_factory.build_rst_stream_frame(
+            stream_id=1,
+            error_code=h2.errors.ErrorCodes.STREAM_CLOSED,
+        ).serialize()
+        assert c.data_to_send() == expected
+
 
 class TestAutomaticFlowControl(object):
     """
diff --git a/test/test_h2_upgrade.py b/test/test_h2_upgrade.py
@@ -260,11 +260,11 @@ def test_cannot_receive_data_stream_1(self, frame_factory):
         )
         c.receive_data(f.serialize())
 
-        expected_frame = frame_factory.build_rst_stream_frame(
+        expected = frame_factory.build_rst_stream_frame(
             stream_id=1,
             error_code=h2.errors.ErrorCodes.STREAM_CLOSED,
-        )
-        assert c.data_to_send() == expected_frame.serialize()
+        ).serialize()
+        assert c.data_to_send() == expected
 
     def test_client_settings_are_applied(self, frame_factory):
         """
diff --git a/test/test_invalid_frame_sequences.py b/test/test_invalid_frame_sequences.py
diff --git a/test/test_stream_reset.py b/test/test_stream_reset.py
diff --git a/test_requirements.txt b/test_requirements.txt
diff --git a/tox.ini b/tox.ini
