This is an automated email from the ASF dual-hosted git repository.
chenBright pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new 92815186 Keep h2 client HPACK state in sync when a stream is refused
(#3630)
92815186 is described below
commit 9281518628bfe98d963c2af03a989dbe3318972c
Author: Weibing Wang <[email protected]>
AuthorDate: Thu Nov 5 14:45:21 2026 +0800
Keep h2 client HPACK state in sync when a stream is refused (#3630)
H2UnsentRequest::AppendAndDestroySelf encoded the request headers
before H2Context::TryToInsertClientStream. Encoding adds the headers
to the dynamic table of the connection, so when the insert failed
(ELIMIT at the server's max_concurrent_streams, EOVERCROWDED, ELOGOFF)
Socket::WriteRequest::Setup dropped the HEADERS while the table kept
the entries. The server's decoder then lags the client's encoder, and
later requests on the connection are decoded with the header fields
of other requests (another :path, another authorization) or fail with
a compression error. ELIMIT and EOVERCROWDED are retried by default on
the same connection, so the retried request itself is decoded wrong.
Insert the stream first, as before 9aa41f79, and encode the headers
only once the request will be sent. The DATA frames the windows allow
are collected on the side and appended after the HEADERS.
Co-authored-by: Mar <[email protected]>
---
src/brpc/policy/http2_rpc_protocol.cpp | 27 +++++---
test/brpc_h2_unsent_message_unittest.cpp | 104 +++++++++++++++++++++++++++++++
2 files changed, 121 insertions(+), 10 deletions(-)
diff --git a/src/brpc/policy/http2_rpc_protocol.cpp
b/src/brpc/policy/http2_rpc_protocol.cpp
index 004430af..eb788d5a 100644
--- a/src/brpc/policy/http2_rpc_protocol.cpp
+++ b/src/brpc/policy/http2_rpc_protocol.cpp
@@ -1864,7 +1864,22 @@ H2UnsentRequest::AppendAndDestroySelf(butil::IOBuf* out,
Socket* socket) {
}
_sctx->Init(ctx, id);
- H2StreamContext* const sctx = _sctx.get();
+ // Insert the stream before encoding the headers and do not fail after
+ // encoding: HPACK encoding adds the headers to the dynamic table of the
+ // connection, and if the request were refused afterwards, the caller
+ // would drop the HEADERS while the table kept the entries, so the
+ // server would decode later requests on this connection with wrong
+ // headers. The DATA that the windows allow now follows the HEADERS.
+ const butil::IOBuf& request_data = _cntl->request_attachment();
+ butil::IOBuf data_frames;
+ const butil::Status insert_status = ctx->TryToInsertClientStream(
+ id, _sctx.get(), request_data, &data_frames);
+ if (!insert_status.ok()) {
+ return insert_status;
+ }
+ _stream_id = id;
+ // TryToInsertClientStream transfers ownership of _sctx to ctx on success.
+ _sctx.release();
HPacker& hpacker = ctx->hpacker();
butil::IOBufAppender appender;
@@ -1889,17 +1904,9 @@ H2UnsentRequest::AppendAndDestroySelf(butil::IOBuf* out,
Socket* socket) {
}
butil::IOBuf frag;
appender.move_to(frag);
- const butil::IOBuf& request_data = _cntl->request_attachment();
PackH2Headers(out, frag, id, remote_settings.max_frame_size,
request_data.empty());
- const butil::Status insert_status =
- ctx->TryToInsertClientStream(id, sctx, request_data, out);
- if (!insert_status.ok()) {
- return insert_status;
- }
- _stream_id = id;
- // TryToInsertClientStream transfers ownership of _sctx to ctx on success.
- _sctx.release();
+ out->append(butil::IOBuf::Movable(data_frames));
const int64_t conn_wu = ctx->ReleaseDeferredWindowUpdate();
if (conn_wu > 0) {
char winbuf[FRAME_HEAD_SIZE + 4];
diff --git a/test/brpc_h2_unsent_message_unittest.cpp
b/test/brpc_h2_unsent_message_unittest.cpp
index a8cd8da2..0d7b39c3 100644
--- a/test/brpc_h2_unsent_message_unittest.cpp
+++ b/test/brpc_h2_unsent_message_unittest.cpp
@@ -19,10 +19,12 @@
// Date: Tue Oct 9 20:27:18 CST 2018
+#include <map>
#include <gflags/gflags.h>
#include <gtest/gtest.h>
#include "bthread/bthread.h"
#include "butil/atomicops.h"
+#include "brpc/details/hpack.h"
#include "brpc/policy/http_rpc_protocol.h"
#include "brpc/policy/http2_rpc_protocol.h"
#include "gperftools_helper.h"
@@ -61,6 +63,44 @@ brpc::policy::H2FrameHead PopFrame(butil::IOBuf* buf,
std::string* payload) {
return frame;
}
+// Decode the header block carried by the HEADERS and CONTINUATION frames in
+// |buf| with |peer|, the HPACK decoder of the remote side.
+// Returns false if the block cannot be decoded.
+bool DecodeHeaderBlock(brpc::HPacker* peer, butil::IOBuf* buf,
+ std::map<std::string, std::string>* headers) {
+ butil::IOBuf block;
+ std::string payload;
+ while (!buf->empty()) {
+ const brpc::policy::H2FrameHead frame = PopFrame(buf, &payload);
+ if (frame.type == brpc::policy::H2_FRAME_HEADERS ||
+ frame.type == brpc::policy::H2_FRAME_CONTINUATION) {
+ block.append(payload);
+ }
+ }
+ headers->clear();
+ while (!block.empty()) {
+ brpc::HPacker::Header h;
+ if (peer->Decode(&block, &h) <= 0) {
+ return false;
+ }
+ (*headers)[h.name] = h.value;
+ }
+ return true;
+}
+
+butil::Status AppendRequest(brpc::Socket* sock, const std::string& uri,
+ const std::string& authorization,
+ butil::IOBuf* out) {
+ brpc::Controller cntl;
+ cntl.http_request().uri() = uri;
+ if (!authorization.empty()) {
+ cntl.http_request().SetHeader("authorization", authorization);
+ }
+ brpc::policy::H2UnsentRequest* request =
+ brpc::policy::H2UnsentRequest::New(&cntl);
+ return request->AppendAndDestroySelf(out, sock);
+}
+
} // namespace
TEST(H2UnsentMessage, split_request_data_by_remote_window) {
@@ -160,6 +200,19 @@ TEST(H2UnsentMessage,
request_does_not_fail_when_body_exceeds_window) {
const butil::Status status =
request->AppendAndDestroySelf(&out, sock.get());
EXPECT_TRUE(status.ok()) << status;
+ // HEADERS (split by max_frame_size) come before the DATA the window
+ // allows, and neither ends the stream.
+ std::string payload;
+ brpc::policy::H2FrameHead frame = PopFrame(&out, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_HEADERS, frame.type);
+ EXPECT_EQ(0, frame.flags & 0x1);
+ while (frame.type != brpc::policy::H2_FRAME_DATA && !out.empty()) {
+ frame = PopFrame(&out, &payload);
+ }
+ EXPECT_EQ(brpc::policy::H2_FRAME_DATA, frame.type);
+ EXPECT_EQ(0, frame.flags & 0x1);
+ EXPECT_EQ("abc", payload);
+ EXPECT_TRUE(out.empty());
brpc::policy::H2StreamContext* sctx = ctx->FindStream(1);
ASSERT_TRUE(sctx != nullptr);
EXPECT_EQ(7u, sctx->_pending_data.size());
@@ -244,6 +297,57 @@ TEST(H2UnsentMessage,
clear_pending_data_releases_overcrowded_buffer) {
EXPECT_EQ(0u, ctx->_pending_data_size);
}
+TEST(H2UnsentMessage,
append_and_destroy_self_keeps_hpack_state_on_refused_insert) {
+ brpc::SocketId id;
+ brpc::SocketUniquePtr sock;
+ brpc::SocketOptions options;
+ options.user = brpc::get_client_side_messenger();
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ ASSERT_EQ(0, brpc::Socket::Address(id, &sock));
+
+ brpc::policy::H2Context* ctx =
+ new brpc::policy::H2Context(sock.get(), nullptr);
+ ASSERT_EQ(0, ctx->Init());
+ sock->initialize_parsing_context(&ctx);
+ ctx->_last_sent_stream_id = 1;
+ ctx->_remote_settings.max_concurrent_streams = 1;
+
+ // The HPACK decoder of the server, which sees only the HEADERS that
+ // were written to the connection.
+ brpc::HPacker peer;
+ ASSERT_EQ(0, peer.Init(ctx->_unack_local_settings.header_table_size));
+ std::map<std::string, std::string> headers;
+
+ butil::IOBuf out;
+ ASSERT_TRUE(AppendRequest(sock.get(), "http://example.com/status";, "",
+ &out).ok());
+ ASSERT_TRUE(DecodeHeaderBlock(&peer, &out, &headers));
+ delete ctx->RemoveStreamAndDeferWU(1);
+
+ ASSERT_TRUE(AppendRequest(sock.get(), "http://example.com/user";,
+ "Bearer alice", &out).ok());
+ ASSERT_TRUE(DecodeHeaderBlock(&peer, &out, &headers));
+ EXPECT_EQ("/user", headers[":path"]);
+ EXPECT_EQ("Bearer alice", headers["authorization"]);
+
+ // Stream 3 takes the only stream the server allows, so this request is
+ // refused and Socket::WriteRequest::Setup drops what it appended.
+ const butil::Status refused = AppendRequest(
+ sock.get(), "http://example.com/user";, "Bearer bob", &out);
+ EXPECT_EQ(brpc::ELIMIT, refused.error_code());
+ EXPECT_EQ(nullptr, ctx->FindStream(5));
+ out.clear();
+
+ // The refused request is retried on the same connection once stream 3
+ // is done. The server must decode the headers that were encoded.
+ delete ctx->RemoveStreamAndDeferWU(3);
+ ASSERT_TRUE(AppendRequest(sock.get(), "http://example.com/user";,
+ "Bearer bob", &out).ok());
+ ASSERT_TRUE(DecodeHeaderBlock(&peer, &out, &headers));
+ EXPECT_EQ("/user", headers[":path"]);
+ EXPECT_EQ("Bearer bob", headers["authorization"]);
+}
+
TEST(H2UnsentMessage, request_throughput) {
brpc::Controller cntl;
butil::IOBuf request_buf;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
