feat core: support websockets over HTTP/2.0 (RFC 8441)

core/functional_tests/CMakeLists.txt

@@ -61,6 +61,9 @@ add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-cache-update)
 add_subdirectory(websocket)
 add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-websocket)
 
+add_subdirectory(websocket_http2)
+add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-websocket-http2)
+
 # WebSocket client requires curl >= 7.86 with WebSocket support
 if(CURL_VERSION_STRING VERSION_GREATER_EQUAL "7.86")
     add_subdirectory(websocket_client)

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/functional_tests/websocket/service.cpp

@@ -146,10 +146,24 @@ class WebsocketsPingPongHandler final : public server::handlers::WebsocketHandle
     }
 };
 
+/// Only configured by the HTTP/2.0 variant of this service, to check what an extended
+/// CONNECT of RFC 8441 aimed at an ordinary handler answers.
+class PlainHandler final : public server::handlers::HttpHandlerBase {
+public:
+    static constexpr std::string_view kName = "plain-handler";
+
+    using HttpHandlerBase::HttpHandlerBase;
+
+    std::string HandleRequestThrow(const server::http::HttpRequest&, server::request::RequestContext&) const override {
+        return "Not a websocket handler";
+    }
+};
+
 int main(int argc, char* argv[]) {
     const auto component_list =
         components::MinimalServerComponentList()
             .Append<WebsocketsHandler>()
+            .Append<PlainHandler>()
             .Append<WebsocketsHandlerAlt>()
             .Append<WebsocketsFullDuplexHandler>()
             .Append<WebsocketsPingPongHandler>()

core/functional_tests/websocket/static_config.yaml

@@ -45,6 +45,10 @@ components_manager:
             task_processor: main-task-processor
             max-remote-payload: 100000
             fragment-size: 10
+        plain-handler:  # Unused here; exercised by the HTTP/2.0 variant of this service.
+            path: /plain
+            method: GET
+            task_processor: main-task-processor
 
         testsuite-support:
 

core/functional_tests/websocket_http2/CMakeLists.txt

@@ -0,0 +1,8 @@
+project(userver-core-tests-websocket-http2 CXX)
+
+# The very same handlers as the HTTP/1.1 websocket test: a websocket handler is not
+# supposed to notice which transport bootstrapped it.
+add_executable(${PROJECT_NAME} "${CMAKE_CURRENT_SOURCE_DIR}/../websocket/service.cpp")
+target_link_libraries(${PROJECT_NAME} userver::core)
+
+userver_chaos_testsuite_add()

core/functional_tests/websocket_http2/static_config.yaml

@@ -0,0 +1,72 @@
+components_manager:
+
+    task_processors:  # Task processor is an executor for coroutine tasks
+        main-task-processor:  # Make a task processor for CPU-bound coroutine tasks.
+            worker_threads: 4  # Process tasks in 4 threads.
+        fs-task-processor:  # Make a separate task processor for filesystem bound tasks.
+            worker_threads: 4
+
+    default_task_processor: main-task-processor
+
+    components:  # Configuring components that were registered via component_list
+        server:
+            listener:  # configuring the main listening socket...
+                connection:
+                    http-version: 2
+                    http2-session:
+                        enable_connect_protocol: true
+                port: 8080  # ...to listen on this port and...
+                task_processor: main-task-processor  # ...process incoming requests on this task processor.
+        logging:
+            fs-task-processor: fs-task-processor
+            loggers:
+                default:
+                    file_path: '@stderr'
+                    level: debug
+                    overflow_behavior: discard  # Drop logs if the system is too busy to write them down.
+
+        websocket-handler:  # Finally! Websocket handler.
+            path: /chat  # Registering handlers '/*' find files.
+            method: GET  # Handle only GET requests.
+            task_processor: main-task-processor  # Run it on CPU bound task processor
+            max-remote-payload: 100000
+            fragment-size: 10
+        websocket-handler-alt:  # Finally! Websocket handler.
+            path: /handler-alt  # Registering handlers '/*' find files.
+            method: GET  # Handle only GET requests.
+            task_processor: main-task-processor  # Run it on CPU bound task processor
+            max-remote-payload: 100000
+            fragment-size: 10
+        websocket-duplex-handler:  # Finally! Websocket handler.
+            path: /duplex  # Registering handlers '/*' find files.
+            method: GET  # Handle only GET requests.
+            task_processor: main-task-processor  # Run it on CPU bound task processor
+            max-remote-payload: 100000
+            fragment-size: 10
+        websocket-ping-pong-handler:
+            path: /ping-pong
+            method: GET
+            task_processor: main-task-processor
+            max-remote-payload: 100000
+            fragment-size: 10
+        plain-handler:  # An ordinary handler, to check that it never accepts a tunnel.
+            path: /plain
+            method: GET
+            task_processor: main-task-processor
+
+        testsuite-support:
+
+        http-client:
+        http-client-core:
+            fs-task-processor: main-task-processor
+        dns-client:
+            fs-task-processor: fs-task-processor
+
+        tests-control:
+            method: POST
+            path: /tests/{action}
+            skip-unregistered-testpoints: true
+            task_processor: main-task-processor
+            testpoint-timeout: 10s
+            testpoint-url: $mockserver/testpoint
+            throttling_enabled: false