From 10b37997ffec64784355e66dcffd3eec6c17ea1a Mon Sep 17 00:00:00 2001 From: SSE4 Date: Mon, 20 Jul 2026 11:32:58 +0700 Subject: [PATCH 1/3] feat core: restore response body streaming for HTTP/2.0 Body streaming over h2 was disabled (SetStreamBody() hit a UINVARIANT whose throw escaped into std::terminate) after the connection refactor dropped the streaming-event wiring from the connection loop, and because the old serial loop head-of-line-blocked concurrent streams. Restore it on top of the concurrent WaitAnyContext loop: * Http2Connection additionally waits on the session's streaming SingleConsumerEvent (NoAutoReset, reset only at the wakeup point where its awaitable has been dropped out of the context) and drains the event queue into the per-stream chunk buffers, resuming deferred nghttp2 data providers. * The response of a streaming handler is submitted upon the first streaming event of its stream (SetHeadersEnd() always precedes the first PushBodyChunk()), so headers go out early while the handler keeps producing; task completion only finalizes stats/access logs. * Events for streams already reset by the client are dropped instead of tearing down the connection; submit errors fail only that request. * An h2c-upgrade first request with a streaming handler (no stream id at SetStreamBody() time, so it used the HTTP/1.1 queue) degrades to a buffered send instead of hanging on a forever-deferred provider. Head-of-line blocking is gone by construction: handler tasks already run concurrently and NGHTTP2_ERR_DEFERRED starves only the chunk-less stream, which nghttp2 skips while serving the other multiplexed streams. Un-skip the three TAXICOMMON-10258 functional tests and add a head-of-line-blocking guard test (slow streamed response must not delay fast requests on the same connection). --- core/functional_tests/http2server/service.cpp | 4 +- .../http2server/tests/test_http2_streaming.py | 46 +++++++++-- core/src/server/http/http2_session.cpp | 33 ++++---- core/src/server/http/http2_session.hpp | 12 ++- core/src/server/http/http2_writer.cpp | 16 +++- core/src/server/http/http_response.cpp | 1 - core/src/server/net/http2_connection.cpp | 81 ++++++++++++++++++- core/src/server/net/http2_connection.hpp | 14 ++++ 8 files changed, 178 insertions(+), 29 deletions(-) diff --git a/core/functional_tests/http2server/service.cpp b/core/functional_tests/http2server/service.cpp index 8fd5dd2da7ec..248d67dfcd44 100644 --- a/core/functional_tests/http2server/service.cpp +++ b/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), {}); diff --git a/core/functional_tests/http2server/tests/test_http2_streaming.py b/core/functional_tests/http2server/tests/test_http2_streaming.py index 101273dc0404..e6bfd9fe55c4 100644 --- a/core/functional_tests/http2server/tests/test_http2_streaming.py +++ b/core/functional_tests/http2server/tests/test_http2_streaming.py @@ -1,11 +1,8 @@ import asyncio -import pytest - DEFAULT_PATH = '/http2server-stream' -@pytest.mark.skip(reason='TAXICOMMON-10258') async def test_body_stream(http2_client, service_client, dynamic_config): part = 'part' count = 100 @@ -25,7 +22,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 +30,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 +39,44 @@ 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_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 diff --git a/core/src/server/http/http2_session.cpp b/core/src/server/http/http2_session.cpp index e1c0d8273b45..3b54ab530235 100644 --- a/core/src/server/http/http2_session.cpp +++ b/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(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(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 diff --git a/core/src/server/http/http2_session.hpp b/core/src/server/http/http2_session.hpp index 3f8ba7deffc6..517f2ecb85c3 100644 --- a/core/src/server/http/http2_session.hpp +++ b/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 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_; }; diff --git a/core/src/server/http/http2_writer.cpp b/core/src/server/http/http2_writer.cpp index 5bf9eedc6e32..3320bd0f75d1 100644 --- a/core/src/server/http/http2_writer.cpp +++ b/core/src/server/http/http2_writer.cpp @@ -105,6 +105,20 @@ class Http2ResponseWriter final { void WriteHttpResponse() { auto data = response_.ExtractData(); + bool buffered_h1_stream = false; + if (response_.IsBodyStreamed() && response_.body_stream_.has_value()) { + // The handler streamed into the HTTP/1.1 queue because the stream + // id was assigned only now, at send time (h2c upgrade). The + // handler has already finished, so buffer the parts and send a + // regular response. + std::string body_part; + while (response_.body_stream_->Pop(body_part)) { + data.append(body_part); + } + response_.body_stream_.reset(); + buffered_h1_stream = true; + } + auto headers = GetHeaders(); const bool is_body_forbidden = IsBodyForbiddenForStatus(response_.status_); @@ -116,7 +130,7 @@ class Http2ResponseWriter final { const auto stream_id = response_.GetStreamId().value(); auto& stream = http2_session_.GetStreamChecked(Stream::Id{stream_id}); - stream.SetStreaming(response_.IsBodyStreamed() && data.empty()); + stream.SetStreaming(response_.IsBodyStreamed() && !buffered_h1_stream && data.empty()); std::size_t bytes = headers.GetSize(); nghttp2_data_provider* provider{nullptr}; diff --git a/core/src/server/http/http_response.cpp b/core/src/server/http/http_response.cpp index 817366707767..18498f007db7 100644 --- a/core/src/server/http/http_response.cpp +++ b/core/src/server/http/http_response.cpp @@ -462,7 +462,6 @@ void SetThrottleReason(http::HttpResponse& http_response, std::string log_reason void HttpResponse::SetStreamBody() { UASSERT(body_stream_producer_.index() == 0); if (GetStreamId().has_value()) { - UINVARIANT(false, "Streaming in HTTP/2.0 is not supported currently."); body_stream_producer_.emplace(GetStreamProducer()); } else { UASSERT(!body_stream_); diff --git a/core/src/server/net/http2_connection.cpp b/core/src/server/net/http2_connection.cpp index a34566d05411..01039077af77 100644 --- a/core/src/server/net/http2_connection.cpp +++ b/core/src/server/net/http2_connection.cpp @@ -27,13 +27,17 @@ constexpr std::string_view kHttp2Preface = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"; constexpr std::string_view kPrefaceBegin = kHttp2Preface.substr(0, kMinLenPrefaceToDetect); constexpr std::uint64_t kSocketId = std::numeric_limits::max(); +constexpr std::uint64_t kStreamingId = std::numeric_limits::max() - 1; -enum class WakeupKind { kSocketReadable, kTaskComputedResponse }; +enum class WakeupKind { kSocketReadable, kStreamingReady, kTaskComputedResponse }; WakeupKind GetWakeupKind(std::uint64_t id) { if (id == kSocketId) { return WakeupKind::kSocketReadable; } + if (id == kStreamingId) { + return WakeupKind::kStreamingReady; + } return WakeupKind::kTaskComputedResponse; } @@ -92,6 +96,7 @@ void Http2Connection::ListenForRequests() { engine::WaitAnyContext wait_any{}; wait_any.Append(kSocketId, GetSocket().GetReadableBase()); + wait_any.Append(kStreamingId, parser_->GetStreamingEvent()); while (!engine::current_task::ShouldCancel()) { StartAllRequestTasks(wait_any); @@ -115,12 +120,23 @@ void Http2Connection::ListenForRequests() { } wait_any.Append(kSocketId, GetSocket().GetReadableBase()); break; + case WakeupKind::kStreamingReady: + // The completed awaitable was dropped out of `wait_any`, so the + // no-auto-reset event has no active awaiter and may be reset + // here. Resetting before the drain keeps a signal arriving + // mid-drain for the next round. `Reset()` is not allowed while + // the event is appended (active awaiter), which is why the + // drain in `OnRequestTaskFinished` leaves the signal alone. + parser_->GetStreamingEvent().Reset(); + HandleStreamingEvents(); + wait_any.Append(kStreamingId, parser_->GetStreamingEvent()); + break; case WakeupKind::kTaskComputedResponse: OnRequestTaskFinished(*ready_id); break; } - UASSERT(wait_any.GetSize() <= config_.http2_session_config.max_concurrent_streams + 1); + UASSERT(wait_any.GetSize() <= config_.http2_session_config.max_concurrent_streams + 2); } } @@ -140,15 +156,69 @@ Http2Connection::RequestTaskContext Http2Connection::StartRequestTask(std::share stats_.active_request_count.Add(1); - return {.task = ConnectionBase::StartRequestTask(request_ptr), .request = std::move(request_ptr)}; + auto task = ConnectionBase::StartRequestTask(request_ptr); + + // `SetStreamBody()` is called synchronously in `StartRequestTask` before + // the handler task is spawned, so `IsBodyStreamed()` is reliable here. + // Requests without a stream id (h2c upgrade) keep the buffered send path. + const auto& response = request_ptr->GetHttpResponse(); + if (response.IsBodyStreamed() && response.GetStreamId().has_value()) { + streamed_requests_.emplace(*response.GetStreamId(), StreamedRequestContext{request_ptr, false}); + } + + return {.task = std::move(task), .request = std::move(request_ptr)}; } void Http2Connection::OnRequestTaskFinished(std::uint64_t event_id) noexcept { - SendResponse(*handler_tasks_[event_id].request); + auto& request = *handler_tasks_[event_id].request; + const auto stream_id = request.GetHttpResponse().GetStreamId(); + if (stream_id.has_value() && streamed_requests_.find(*stream_id) != streamed_requests_.end()) { + // Drain the remaining body parts. `ResponseBodyStream` always pushes a + // final event before the handler task completes, so this also submits + // the response if no streaming event was processed for it yet. + try { + HandleStreamingEvents(); + } catch (const std::exception& ex) { + LOG_ERROR() << "Error while sending streamed body parts: " << ex; + request.GetHttpResponse().SetSendFailed(std::chrono::steady_clock::now()); + } + SubmitStreamedResponseIfPending(*stream_id); + FinalizeResponse(request); + streamed_requests_.erase(*stream_id); + } else { + SendResponse(request); + } handler_tasks_.erase(event_id); } +void Http2Connection::HandleStreamingEvents() { + http::impl::Http2StreamEvent event; + while (parser_->PopStreamingEventNoblock(event)) { + // The first event for a stream means its headers are complete + // (`SetHeadersEnd()` precedes the first `PushBodyChunk()`), so the + // response with its deferred body provider is submitted here. + SubmitStreamedResponseIfPending(event.stream_id); + parser_->ApplyStreamingEvent(std::move(event)); + event = {}; + } + parser_->WriteWhileWant(); +} + +void Http2Connection::SubmitStreamedResponseIfPending(std::int32_t stream_id) noexcept { + const auto it = streamed_requests_.find(stream_id); + if (it == streamed_requests_.end() || it->second.submit_attempted) { + return; + } + it->second.submit_attempted = true; + SubmitResponse(*it->second.request); +} + void Http2Connection::SendResponse(http::HttpRequest& request) noexcept { + SubmitResponse(request); + FinalizeResponse(request); +} + +void Http2Connection::SubmitResponse(http::HttpRequest& request) noexcept { auto& response = request.GetHttpResponse(); UASSERT(!response.IsSent()); request.SetStartSendResponseTime(); @@ -175,6 +245,9 @@ void Http2Connection::SendResponse(http::HttpRequest& request) noexcept { } else { response.SetSendFailed(std::chrono::steady_clock::now()); } +} + +void Http2Connection::FinalizeResponse(http::HttpRequest& request) noexcept { request.SetFinishSendResponseTime(); stats_.active_request_count.Subtract(1); ++stats_.requests_processed_count; diff --git a/core/src/server/net/http2_connection.hpp b/core/src/server/net/http2_connection.hpp index e6c2d5addad0..87158e99f716 100644 --- a/core/src/server/net/http2_connection.hpp +++ b/core/src/server/net/http2_connection.hpp @@ -3,6 +3,7 @@ #include #include #include +#include #include #include @@ -59,11 +60,23 @@ class Http2Connection final : public ConnectionBase { HttpRequestPtr request; }; + // A request whose handler streams the response body (`IsBodyStreamed()`). + // Its response is submitted upon the first streaming event instead of on + // handler task completion. + struct StreamedRequestContext final { + HttpRequestPtr request; + bool submit_attempted{false}; + }; + void ListenForRequests(); RequestTaskContext StartRequestTask(std::shared_ptr&& request_ptr) noexcept; void StartAllRequestTasks(engine::WaitAnyContext& wait_any); void OnRequestTaskFinished(std::uint64_t event_id) noexcept; + void HandleStreamingEvents(); + void SubmitStreamedResponseIfPending(std::int32_t stream_id) noexcept; void SendResponse(http::HttpRequest& request) noexcept; + void SubmitResponse(http::HttpRequest& request) noexcept; + void FinalizeResponse(http::HttpRequest& request) noexcept; std::unique_ptr MakeParser(); void EnsureHttp2(); @@ -82,6 +95,7 @@ class Http2Connection final : public ConnectionBase { engine::io::Sockaddr remote_address_; std::unique_ptr parser_; utils::SlotMap handler_tasks_; + std::unordered_map streamed_requests_; }; } // namespace server::net From a76271d5e7757b77d9bc6046f8cf944dcdd900ae Mon Sep 17 00:00:00 2001 From: SSE4 Date: Mon, 20 Jul 2026 13:41:20 +0700 Subject: [PATCH 2/3] test core: harden HTTP/2.0 body streaming coverage; fix debug assert race New functional tests for the restored h2 streaming: * client reset mid-stream: the handler keeps producing after RST_STREAM; its events must be dropped and the connection must stay usable; * h2c upgrade whose first response is streamed: must degrade to a buffered send, not hang; * flow-control backpressure: a 100 KiB streamed body behind a 1 KiB stream window (and past the connection window) must arrive intact, WINDOW_UPDATE-gated. New unit tests (there were none for the streaming machinery): event queue FIFO/signal contract, WaitAny compatibility of the streaming event, dropped events for unknown streams, and SetStreamBody() picking the h2 producer (previously a process-fatal UINVARIANT). Under TSan the old PushBodyChunk() assert read the response data while the connection coroutine's ExtractData() moved it (the response is now submitted concurrently with the producing handler). Check the data only before the first chunk is announced - after that SetBody() is already impossible and asserts in SetBody() itself. --- .../http2server/tests/test_http2_streaming.py | 132 ++++++++++++++++++ core/src/server/http/http2_session_test.cpp | 78 +++++++++++ .../server/http/http_response_body_stream.cpp | 9 +- 3 files changed, 218 insertions(+), 1 deletion(-) diff --git a/core/functional_tests/http2server/tests/test_http2_streaming.py b/core/functional_tests/http2server/tests/test_http2_streaming.py index e6bfd9fe55c4..73bba45f967a 100644 --- a/core/functional_tests/http2server/tests/test_http2_streaming.py +++ b/core/functional_tests/http2server/tests/test_http2_streaming.py @@ -1,8 +1,23 @@ import asyncio +import h2.connection +import h2.events +import h2.settings + +import utils + DEFAULT_PATH = '/http2server-stream' +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 @@ -80,3 +95,120 @@ async def test_body_stream_no_head_of_line_blocking( 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 diff --git a/core/src/server/http/http2_session_test.cpp b/core/src/server/http/http2_session_test.cpp index c4a40812cae3..353c1cb882d7 100644 --- a/core/src/server/http/http2_session_test.cpp +++ b/core/src/server/http/http2_session_test.cpp @@ -11,6 +11,7 @@ #include #include #include +#include #include #include @@ -382,6 +383,83 @@ UTEST_F(Http2SessionTest, ForCurl) { EXPECT_EQ(request->GetMethod(), HttpMethod::kPost); } +UTEST(Http2SessionStreaming, EventQueueIsFifoPerStreamAndSignals) { + const auto queue = impl::Http2StreamEventQueue::Create(); + engine::SingleConsumerEvent event{engine::SingleConsumerEvent::NoAutoReset()}; + impl::Http2StreamEventProducer producer{*queue, event}; + auto consumer = queue->GetConsumer(); + + producer.PushEvent({1, "first"}); + producer.PushEvent({1, "second"}); + producer.CloseStream(1); + EXPECT_TRUE(event.IsReady()); + + impl::Http2StreamEvent popped; + ASSERT_TRUE(consumer.PopNoblock(popped)); + EXPECT_EQ(popped.stream_id, 1); + EXPECT_EQ(popped.body_part, "first"); + EXPECT_FALSE(popped.is_end); + ASSERT_TRUE(consumer.PopNoblock(popped)); + EXPECT_EQ(popped.body_part, "second"); + ASSERT_TRUE(consumer.PopNoblock(popped)); + EXPECT_TRUE(popped.is_end); + EXPECT_FALSE(consumer.PopNoblock(popped)); +} + +UTEST(Http2SessionStreaming, StreamingEventIsWaitAnyCompatible) { + auto parser = CreateTestParser([](std::shared_ptr&&) {}, USERVER_NAMESPACE::http::HttpVersion::k2); + auto& session = dynamic_cast(*parser); + + auto& event = session.GetStreamingEvent(); + EXPECT_FALSE(event.IsAutoReset()); + // Auto-reset events UINVARIANT-abort here; the connection loop appends + // this token to its WaitAnyContext. + EXPECT_FALSE(event.GetAwaitableToken().IsEmpty()); +} + +UTEST(Http2SessionStreaming, EventForUnknownStreamIsDropped) { + auto parser = CreateTestParser([](std::shared_ptr&&) {}, USERVER_NAMESPACE::http::HttpVersion::k2); + auto& session = dynamic_cast(*parser); + + impl::Http2StreamEvent event; + EXPECT_FALSE(session.PopStreamingEventNoblock(event)); + + // A handler may still be producing body parts after the client reset the + // stream; such events must be dropped, not tear down the connection. + impl::Http2StreamEvent late{42, "late chunk", true}; + EXPECT_NO_THROW(session.ApplyStreamingEvent(std::move(late))); +} + +UTEST(Http2SessionStreaming, SetStreamBodyPicksHttp2Producer) { + const auto queue = impl::Http2StreamEventQueue::Create(); + engine::SingleConsumerEvent event{engine::SingleConsumerEvent::NoAutoReset()}; + + request::ResponseDataAccounter accounter; + const auto request = HttpRequestBuilder{accounter} + .SetMethod(HttpMethod::kGet) + .SetHttpMajor(2) + .SetHttpMinor(0) + .SetUrl("/") + .SetResponseStreamId(1) + .SetStreamProducer(impl::Http2StreamEventProducer{*queue, event}) + .Build(); + auto& response = request->GetHttpResponse(); + + // Used to UINVARIANT-abort the whole process for HTTP/2 responses. + response.SetStreamBody(); + EXPECT_TRUE(response.IsBodyStreamed()); + + auto producer = response.GetBodyProducer(); + ASSERT_TRUE(std::holds_alternative(producer)); + + auto consumer = queue->GetConsumer(); + std::get(producer).PushEvent({1, "chunk"}); + impl::Http2StreamEvent popped; + ASSERT_TRUE(consumer.PopNoblock(popped)); + EXPECT_EQ(popped.body_part, "chunk"); + EXPECT_TRUE(event.IsReady()); +} + } // namespace server::http USERVER_NAMESPACE_END diff --git a/core/src/server/http/http_response_body_stream.cpp b/core/src/server/http/http_response_body_stream.cpp index de2f5c3d6cb4..e7ded0820ade 100644 --- a/core/src/server/http/http_response_body_stream.cpp +++ b/core/src/server/http/http_response_body_stream.cpp @@ -24,7 +24,14 @@ ResponseBodyStream::~ResponseBodyStream() { void ResponseBodyStream::PushBodyChunk(std::string&& chunk, engine::Deadline deadline) { UASSERT_MSG(headers_ended_, "SetEndOfHeaders() was not called before PushBodyChunk()"); - UASSERT_MSG(http_response_.GetData().empty(), "PushBodyChunk() was called after SetBody()"); + // Only check before the first chunk is announced: after that the + // connection coroutine may be sending the response concurrently and reads + // of the response data would race with it. SetBody() after the first + // chunk is already asserted in SetBody() itself. + UASSERT_MSG( + headers_end_sent_ || http_response_.GetData().empty(), + "PushBodyChunk() was called after SetBody()" + ); if (headers_ended_ && !headers_end_sent_) { http_response_.SetHeadersEnd(); From 277674f7a3d765ca10c607ca2455c35e3f934b84 Mon Sep 17 00:00:00 2001 From: SSE4 Date: Mon, 20 Jul 2026 14:10:50 +0700 Subject: [PATCH 3/3] test core: verify streamed body integrity across multiplexed streams The concurrent streaming tests used identical payloads on every stream, so bytes leaking between concurrently multiplexed streams (or chunks reordered within one stream) could not be detected. Echo a payload whose every block encodes the stream number and block position, 20 streams multiplexed on one connection, and require byte-exact responses. --- .../http2server/tests/test_http2_streaming.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/core/functional_tests/http2server/tests/test_http2_streaming.py b/core/functional_tests/http2server/tests/test_http2_streaming.py index 73bba45f967a..885bb53ab5b0 100644 --- a/core/functional_tests/http2server/tests/test_http2_streaming.py +++ b/core/functional_tests/http2server/tests/test_http2_streaming.py @@ -56,6 +56,24 @@ async def test_body_stream_concurrent( 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,