feat core: restore response body streaming for HTTP/2.0

core/functional_tests/http2server/service.cpp

@@ -53,13 +53,15 @@ class HandlerHttp2Stream final : public server::handlers::HttpHandlerBase {
             const auto& count_str = req.GetArg("count");
             const std::size_t count = std::stoi(count_str);
             UASSERT(count != 0);
+            const auto& delay_str = req.GetArg("delay_ms");
+            const std::chrono::milliseconds delay{delay_str.empty() ? 2 : std::stoi(delay_str)};
             stream.SetStatusCode(200);
             stream.SetEndOfHeaders();
             for (std::size_t i = 0; i < count - 1; i++) {
                 std::string part{body_part};
                 stream.PushBodyChunk(std::move(part), {});
                 // Some pause...
-                engine::SleepFor(std::chrono::milliseconds{2});
+                engine::SleepFor(delay);
             }
             std::string part{body_part};
             stream.PushBodyChunk(std::move(part), {});

core/functional_tests/http2server/tests/test_http2_streaming.py

@@ -1,11 +1,23 @@
 import asyncio
 
-import pytest
+import h2.connection
+import h2.events
+import h2.settings
+
+import utils
 
 DEFAULT_PATH = '/http2server-stream'
 
 
-@pytest.mark.skip(reason='TAXICOMMON-10258')
+def _stream_headers(query: str) -> list:
+    return [
+        (':method', 'GET'),
+        (':path', f'{DEFAULT_PATH}?{query}'),
+        (':scheme', 'http'),
+        (':authority', 'localhost'),
+    ]
+
+
 async def test_body_stream(http2_client, service_client, dynamic_config):
     part = 'part'
     count = 100
@@ -25,7 +37,6 @@ async def _stream_request(client, req_per_client):
         assert data == r.text
 
 
-@pytest.mark.skip(reason='TAXICOMMON-10258')
 async def test_body_stream_small_pieces(
     http2_client,
     service_client,
@@ -34,7 +45,6 @@ async def test_body_stream_small_pieces(
     await _stream_request(http2_client, 1)
 
 
-@pytest.mark.skip(reason='TAXICOMMON-10258')
 async def test_body_stream_concurrent(
     http2_client,
     service_client,
@@ -44,3 +54,179 @@ async def test_body_stream_concurrent(
     req_per_client = 10
     tasks = [_stream_request(http2_client, req_per_client) for _ in range(clients_count)]
     await asyncio.gather(*tasks)
+
+
+async def test_body_stream_concurrent_unique_bodies(
+    http2_client,
+    service_client,
+    dynamic_config,
+):
+    # Each stream echoes a payload whose every 8-byte block encodes the
+    # stream number and the block position. Unlike identical payloads, this
+    # detects bytes leaking between concurrently multiplexed streams as well
+    # as chunk reordering within one stream.
+    async def echo_stream(i):
+        data = ''.join(f'{i:02d}:{j:04d};' for j in range(128))  # 1 KiB
+        r = await http2_client.get(DEFAULT_PATH, params={'type': 'ne'}, data=data)
+        assert 200 == r.status_code
+        assert data == r.text
+
+    await asyncio.gather(*[echo_stream(i) for i in range(20)])
+
+
+async def test_body_stream_no_head_of_line_blocking(
+    http2_client,
+    service_client,
+    dynamic_config,
+):
+    # A slow streamed response (~5s) on one stream must not delay other
+    # requests multiplexed on the same connection. If it did, each "fast"
+    # request below would complete only after the slow stream finishes and
+    # trip its timeout.
+    part = 'x'
+    count = 50
+    slow = asyncio.create_task(
+        http2_client.get(
+            DEFAULT_PATH,
+            params={
+                'type': 'eq',
+                'body_part': part,
+                'count': count,
+                'delay_ms': 100,
+            },
+            timeout=30.0,
+        ),
+    )
+    try:
+        for _ in range(5):
+            r = await asyncio.wait_for(
+                http2_client.get(
+                    '/http2server',
+                    params={'type': 'echo-body'},
+                    data='ping',
+                ),
+                timeout=2.0,
+            )
+            assert 200 == r.status_code
+            assert 'ping' == r.text
+    finally:
+        r = await slow
+    assert 200 == r.status_code
+    assert part * count == r.text
+
+
+async def test_reset_mid_stream_keeps_connection_usable(
+    create_connection,
+    service_client,
+):
+    async with create_connection() as (sock, conn):
+        # A slow stream: the handler will keep producing for ~3s after the
+        # client resets the stream; those events must be dropped, not tear
+        # down the connection or the process.
+        stream_id = conn.get_next_available_stream_id()
+        conn.send_headers(
+            stream_id,
+            _stream_headers('type=eq&body_part=part&count=30&delay_ms=100'),
+            end_stream=True,
+        )
+        await sock.sendall(conn.data_to_send())
+
+        events = []
+        while not any(isinstance(event, h2.events.DataReceived) for event in events):
+            events += await utils.send_and_receive(sock, conn)
+
+        conn.reset_stream(stream_id, error_code=0x8)  # CANCEL
+        await sock.sendall(conn.data_to_send())
+
+        # The same connection must still serve requests, concurrently with
+        # the handler of the reset stream still pushing body parts.
+        echo_stream_id = conn.get_next_available_stream_id()
+        conn.send_headers(
+            echo_stream_id,
+            [
+                (':method', 'GET'),
+                (':path', '/http2server?type=echo-header'),
+                (':scheme', 'http'),
+                (':authority', 'localhost'),
+                ('echo-header', 'still-alive'),
+            ],
+            end_stream=True,
+        )
+        await sock.sendall(conn.data_to_send())
+
+        events = await utils.receive_until_stream_ended(sock, conn)
+        assert b'still-alive' == utils.response_data(events)
+
+
+async def test_h2c_upgrade_with_streamed_response(create_socket, service_client):
+    # The first request of an h2c upgrade is parsed as HTTP/1.1, so the
+    # streamed response has no HTTP/2 producer; it must degrade to a
+    # buffered send instead of hanging on a forever-deferred provider.
+    async with create_socket() as sock:
+        conn = h2.connection.H2Connection()
+        settings_header = conn.initiate_upgrade_connection().decode('ascii')
+        request = (
+            f'GET {DEFAULT_PATH}?type=eq&body_part=part&count=10 HTTP/1.1\r\n'
+            'Host: localhost\r\n'
+            'Connection: Upgrade, HTTP2-Settings\r\n'
+            'Upgrade: h2c\r\n'
+            f'HTTP2-Settings: {settings_header}\r\n'
+            '\r\n'
+        )
+        await sock.sendall(request.encode('ascii'))
+
+        receive = b''
+        while utils.HTTP1_HEADERS_END not in receive:
+            receive += await sock.recv(utils.RECEIVE_SIZE)
+        headers, _, http2_data = receive.partition(utils.HTTP1_HEADERS_END)
+        assert headers.startswith(b'HTTP/1.1 101 Switching Protocols')
+
+        events = conn.receive_data(http2_data) if http2_data else []
+        await sock.sendall(conn.data_to_send())
+        while not any(isinstance(event, h2.events.StreamEnded) for event in events):
+            events += await utils.send_and_receive(sock, conn)
+
+        assert b'part' * 10 == utils.response_data(events)
+
+
+async def test_flow_control_backpressure(create_connection, service_client):
+    # With a tiny stream window the server may only produce as fast as the
+    # client opens the window with WINDOW_UPDATEs; the deferred provider must
+    # resume each time instead of stalling or flooding.
+    window = 1024
+    part = 'x' * 1024
+    count = 100  # total body is 100 KiB, also exceeds the connection window
+
+    async with create_connection() as (sock, conn):
+        conn.update_settings(
+            {h2.settings.SettingCodes.INITIAL_WINDOW_SIZE: window},
+        )
+        stream_id = conn.get_next_available_stream_id()
+        conn.send_headers(
+            stream_id,
+            _stream_headers(f'type=eq&body_part={part}&count={count}&delay_ms=0'),
+            end_stream=True,
+        )
+        await sock.sendall(conn.data_to_send())
+
+        body = b''
+        ended = False
+        while not ended:
+            receive = await sock.recv(utils.RECEIVE_SIZE)
+            if not receive:
+                raise RuntimeError('Socket connection was closed by the other side')
+            for event in conn.receive_data(receive):
+                if isinstance(event, h2.events.DataReceived):
+                    body += event.data
+                    conn.acknowledge_received_data(
+                        event.flow_controlled_length,
+                        event.stream_id,
+                    )
+                elif isinstance(event, h2.events.StreamEnded):
+                    ended = True
+            data = conn.data_to_send()
+            if data:
+                await sock.sendall(data)
+
+        assert len(body) == len(part) * count
+        assert part.encode() * count == body

core/src/server/http/http2_session.cpp

@@ -58,7 +58,6 @@ Http2Session::Http2Session(
       streaming_consumer_(streaming_queue_->GetConsumer())
 {
     UASSERT(streaming_queue_);
-    UASSERT(streaming_event_.IsAutoReset());
 
     nghttp2_session_callbacks* callbacks{nullptr};
     UINVARIANT(nghttp2_session_callbacks_new(&callbacks) == 0, "Failed to init callbacks for HTTP/2.0");
@@ -367,21 +366,25 @@ void Http2Session::WriteWhileWant() {
 
 engine::SingleConsumerEvent& Http2Session::GetStreamingEvent() { return streaming_event_; }
 
-void Http2Session::HandleStreamingEvents() {
-    impl::Http2StreamEvent event;
-    while (streaming_consumer_.PopNoblock(event)) {
-        UASSERT(event.stream_id != -1);
-        auto& stream = GetStreamChecked(Stream::Id{event.stream_id});
-        if (stream.IsDeferred()) {
-            const auto res = nghttp2_session_resume_data(session_.get(), static_cast<std::int32_t>(stream.GetId()));
-            ThrowIfErr(res, "Error while resume_data");
-            stream.SetDeferred(false);
-        }
-        stream.PushChunk(std::move(event.body_part));
-        stream.SetEnd(event.is_end);
-        event = {};
+bool Http2Session::PopStreamingEventNoblock(impl::Http2StreamEvent& event) {
+    return streaming_consumer_.PopNoblock(event);
+}
+
+void Http2Session::ApplyStreamingEvent(impl::Http2StreamEvent&& event) {
+    UASSERT(event.stream_id != -1);
+    auto* stream = static_cast<Stream*>(nghttp2_session_get_stream_user_data(session_.get(), event.stream_id));
+    if (stream == nullptr) {
+        // The stream is already closed (e.g. reset by the client) while the
+        // handler was still producing body parts. Drop the event.
+        return;
+    }
+    if (stream->IsDeferred()) {
+        const auto res = nghttp2_session_resume_data(session_.get(), event.stream_id);
+        ThrowIfErr(res, "Error while resume_data");
+        stream->SetDeferred(false);
     }
-    WriteWhileWant();
+    stream->PushChunk(std::move(event.body_part));
+    stream->SetEnd(event.is_end);
 }
 
 }  // namespace server::http

core/src/server/http/http2_session.hpp

@@ -62,7 +62,13 @@ class Http2Session final : public request::RequestParser {
     engine::SingleConsumerEvent& GetStreamingEvent();
 
     void WriteWhileWant();
-    void HandleStreamingEvents();
+
+    // Returns false if there are no pending streaming events.
+    [[nodiscard]] bool PopStreamingEventNoblock(impl::Http2StreamEvent& event);
+
+    // Applies a body-streaming event to its stream. Events for streams that
+    // are already closed (e.g. reset by the client) are dropped.
+    void ApplyStreamingEvent(impl::Http2StreamEvent&& event);
 
     bool ConnectionIsOk() const;
 
@@ -131,7 +137,9 @@ class Http2Session final : public request::RequestParser {
     engine::io::RwBase* socket_;
 
     std::shared_ptr<impl::Http2StreamEventQueue> streaming_queue_{nullptr};
-    engine::SingleConsumerEvent streaming_event_;
+    // No-auto-reset: the event is awaited through WaitAny in the connection
+    // loop, which resets it manually before draining the queue.
+    engine::SingleConsumerEvent streaming_event_{engine::SingleConsumerEvent::NoAutoReset{}};
     impl::Http2StreamEventQueue::Consumer streaming_consumer_;
 };