From de135def40f8237f260f3b922813894a69929858 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Mon, 13 Jul 2026 17:42:44 -0400 Subject: [PATCH 1/7] DRIVER-409 Add request compression support --- CMakeLists.txt | 2 +- README.md | 20 ++++ include/scylladb/alternator/config.h | 17 +++ src/aws_dynamodb_helper.cpp | 72 +++++++++++- src/config.cpp | 1 + src/http_compression.cpp | 85 +++++++++++++- src/http_compression.h | 2 + tests/aws_dynamodb_helper_test.cpp | 160 +++++++++++++++++++++++++++ tests/http_client_test.cpp | 13 +++ 9 files changed, 368 insertions(+), 4 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index a2da241..761210c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -54,7 +54,7 @@ if(ZLIB_FOUND) target_link_libraries(alternator_client_cpp PUBLIC ZLIB::ZLIB) target_compile_definitions(alternator_client_cpp PUBLIC SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB=1) else() - message(STATUS "ZLIB not found; gzip/deflate response decoding requires a caller-provided HttpContentEncodingDecoder") + message(STATUS "ZLIB not found; gzip request encoding and gzip/deflate response decoding are unavailable") endif() if(CURL_FOUND) diff --git a/README.md b/README.md index ecc5f8e..6b73a6a 100644 --- a/README.md +++ b/README.md @@ -140,6 +140,25 @@ Header optimization applies only to requests routed to known Alternator endpoints through the factory installed by `DynamoDBHelper::ApplyToSDKOptions()` before `Aws::InitAPI()`. +### HTTP Request Compression + +Request compression is disabled by default. Configure a request content encoder +before constructing `DynamoDBHelper`, then install the Alternator HTTP client +factory with `DynamoDBHelper::ApplyToSDKOptions()` before `Aws::InitAPI()`. +The factory compresses request bodies only for requests routed to known +Alternator nodes and sets the matching `Content-Encoding` and `Content-Length` +headers before sending the request. + +```cpp +scylladb::alternator::Config cfg; +cfg.content_encoding_encoder = + std::make_shared(); +``` + +Only gzip request compression is built in. `GzipContentEncodingEncoder` requires +zlib at build time. The `HttpContentEncodingEncoder` interface keeps request +compression extensible for other content encodings. + ### HTTP Response Compression Response compression is disabled by default. When configured, the built-in @@ -378,6 +397,7 @@ auto batch_plan = helper.NewBatchWriteQueryPlan({ - Reused libcurl discovery HTTP connections with an opt-out switch. - TLS session cache enable/disable, cache size, and timeout configuration for HTTPS discovery. - Persistent AWS SDK DynamoDB HTTP connection pooling via `max_connections`. +- Optional gzip request compression for AWS SDK DynamoDB requests routed through the Alternator HTTP client factory. - Active, quarantined, and down node pools with rigid observation-based transitions. - Round-robin `NextNode()` and flat per-request query plans, including deterministic seeded plans for affinity callers. - Key-route affinity helpers for single-write partition keys and batch-write preferred-node voting. diff --git a/include/scylladb/alternator/config.h b/include/scylladb/alternator/config.h index b7e43e4..7795b10 100644 --- a/include/scylladb/alternator/config.h +++ b/include/scylladb/alternator/config.h @@ -25,6 +25,14 @@ class HttpContentEncodingDecoder { [[nodiscard]] virtual std::string Decode(std::string body, const std::string& content_encoding) const = 0; }; +class HttpContentEncodingEncoder { +public: + virtual ~HttpContentEncodingEncoder() = default; + + [[nodiscard]] virtual std::string ContentEncoding() const = 0; + [[nodiscard]] virtual std::string Encode(std::string body) const = 0; +}; + class ZlibContentEncodingDecoder final : public HttpContentEncodingDecoder { public: explicit ZlibContentEncodingDecoder(std::vector accepted_response_encodings = {"gzip", "deflate"}); @@ -36,6 +44,14 @@ class ZlibContentEncodingDecoder final : public HttpContentEncodingDecoder { std::vector accepted_response_encodings_; }; +class GzipContentEncodingEncoder final : public HttpContentEncodingEncoder { +public: + GzipContentEncodingEncoder(); + + [[nodiscard]] std::string ContentEncoding() const override; + [[nodiscard]] std::string Encode(std::string body) const override; +}; + struct HeaderOptimizationContext { bool credentials_configured = false; bool user_agent_configured = false; @@ -87,6 +103,7 @@ struct Config { unsigned max_connections = 100; bool reuse_discovery_connections = true; + std::shared_ptr content_encoding_encoder; std::vector> content_encoding_decoders; std::string user_agent = "scylladb-alternator-client-cpp/devel"; std::shared_ptr header_optimization; diff --git a/src/aws_dynamodb_helper.cpp b/src/aws_dynamodb_helper.cpp index 3c8c777..a937739 100644 --- a/src/aws_dynamodb_helper.cpp +++ b/src/aws_dynamodb_helper.cpp @@ -13,6 +13,7 @@ #include #include #include +#include #include #include #include @@ -125,6 +126,15 @@ std::string ReadResponseBody(Aws::IOStream& body) { return std::string(std::istreambuf_iterator(body), std::istreambuf_iterator()); } +std::string ReadRequestBody(Aws::IOStream& body) { + body.clear(); + body.seekg(0, std::ios::beg); + auto value = std::string(std::istreambuf_iterator(body), std::istreambuf_iterator()); + body.clear(); + body.seekg(0, std::ios::beg); + return value; +} + void WriteResponseBody(Aws::IOStream& body, const std::string& value) { if (!value.empty()) { body.write(value.data(), static_cast(value.size())); @@ -134,6 +144,34 @@ void WriteResponseBody(Aws::IOStream& body, const std::string& value) { body.seekg(0, std::ios::beg); } +std::shared_ptr NewRequestBodyStream(const std::string& value) { + auto stream = Aws::MakeShared(kAllocationTag); + if (!value.empty()) { + stream->write(value.data(), static_cast(value.size())); + } + stream->flush(); + stream->clear(); + stream->seekg(0, std::ios::beg); + return stream; +} + +void EncodeAwsRequestBody( + const std::shared_ptr& request, + const std::shared_ptr& content_encoding_encoder, + const std::string& content_encoding_value) { + if (!request || !content_encoding_encoder || content_encoding_value.empty() || + !request->GetContentBody() || request->IsEventStreamRequest() || + request->HasTransferEncoding() || request->HasContentEncoding()) { + return; + } + + const auto encoded_body = content_encoding_encoder->Encode(ReadRequestBody(*request->GetContentBody())); + request->AddContentBody(NewRequestBodyStream(encoded_body)); + request->SetContentEncoding(Aws::String(content_encoding_value.c_str())); + request->SetContentLength(Aws::String(std::to_string(encoded_body.size()).c_str())); + request->DeleteHeader("x-amz-content-sha256"); +} + std::shared_ptr DecodeCompressedAwsResponse( const std::shared_ptr& request, const std::shared_ptr& response, @@ -211,6 +249,12 @@ HeaderOptimizationPolicy BuildHeaderOptimizationPolicy(const Config& config) { const auto headers = config.header_optimization->AllowedHeaders( HeaderOptimizationContextFromConfig(config)); + if (config.content_encoding_encoder) { + auto with_compression_headers = headers; + with_compression_headers.push_back(Aws::Http::CONTENT_ENCODING_HEADER); + with_compression_headers.push_back(Aws::Http::CONTENT_LENGTH_HEADER); + return {true, NormalizeHeaderAllowlist(with_compression_headers)}; + } return {true, NormalizeHeaderAllowlist(headers)}; } @@ -252,11 +296,14 @@ class AlternatorHttpClient final : public Aws::Http::HttpClient { public: AlternatorHttpClient(std::shared_ptr nodes, std::uint16_t endpoint_override_port, + std::shared_ptr content_encoding_encoder, std::vector> content_encoding_decoders, HeaderOptimizationPolicy header_optimization, std::shared_ptr delegate) : nodes_(std::move(nodes)) , endpoint_override_port_(endpoint_override_port) + , content_encoding_encoder_(std::move(content_encoding_encoder)) + , content_encoding_value_(detail::BuildContentEncodingValue(content_encoding_encoder_)) , content_encoding_decoders_(std::move(content_encoding_decoders)) , accept_encoding_value_(detail::BuildAcceptEncodingValue(content_encoding_decoders_)) , header_optimization_(std::move(header_optimization)) @@ -278,6 +325,7 @@ class AlternatorHttpClient final : public Aws::Http::HttpClient { request->SetAcceptEncoding(Aws::String(accept_encoding_value_.c_str())); } if (!node.Empty()) { + EncodeAwsRequestBody(request, content_encoding_encoder_, content_encoding_value_); OptimizeRequestHeaders(request, header_optimization_); } @@ -301,6 +349,8 @@ class AlternatorHttpClient final : public Aws::Http::HttpClient { private: std::shared_ptr nodes_; std::uint16_t endpoint_override_port_ = 0; + std::shared_ptr content_encoding_encoder_; + std::string content_encoding_value_; std::vector> content_encoding_decoders_; std::string accept_encoding_value_; HeaderOptimizationPolicy header_optimization_; @@ -311,10 +361,12 @@ class AlternatorHttpClientFactory final : public Aws::Http::HttpClientFactory { public: AlternatorHttpClientFactory(std::shared_ptr nodes, std::uint16_t endpoint_override_port, + std::shared_ptr content_encoding_encoder, std::vector> content_encoding_decoders, HeaderOptimizationPolicy header_optimization) : nodes_(std::move(nodes)) , endpoint_override_port_(endpoint_override_port) + , content_encoding_encoder_(std::move(content_encoding_encoder)) , content_encoding_decoders_(std::move(content_encoding_decoders)) , header_optimization_(std::move(header_optimization)) { if (!nodes_) { @@ -333,6 +385,7 @@ class AlternatorHttpClientFactory final : public Aws::Http::HttpClientFactory { kAllocationTag, nodes_, endpoint_override_port_, + content_encoding_encoder_, content_encoding_decoders_, header_optimization_, std::move(delegate)); @@ -368,6 +421,7 @@ class AlternatorHttpClientFactory final : public Aws::Http::HttpClientFactory { private: std::shared_ptr nodes_; std::uint16_t endpoint_override_port_ = 0; + std::shared_ptr content_encoding_encoder_; std::vector> content_encoding_decoders_; HeaderOptimizationPolicy header_optimization_; }; @@ -379,12 +433,14 @@ std::string EndpointOverride(std::uint16_t port, const std::string& scheme) { std::shared_ptr NewAlternatorHttpClientFactory( std::shared_ptr nodes, std::uint16_t endpoint_override_port, + std::shared_ptr content_encoding_encoder, std::vector> content_encoding_decoders, HeaderOptimizationPolicy header_optimization) { return Aws::MakeShared( kAllocationTag, std::move(nodes), endpoint_override_port, + std::move(content_encoding_encoder), std::move(content_encoding_decoders), std::move(header_optimization)); } @@ -798,6 +854,7 @@ std::shared_ptr DynamoDBHelper::NewHttpClientFacto return NewAlternatorHttpClientFactory( nodes_, config_.port, + config_.content_encoding_encoder, config_.content_encoding_decoders, BuildHeaderOptimizationPolicy(config_)); } @@ -829,10 +886,21 @@ std::shared_ptr DynamoDBHelper::Nodes() const { void DynamoDBHelper::ApplyToSDKOptions(Aws::SDKOptions& options) const { auto nodes = nodes_; const auto port = config_.port; + const auto content_encoding_encoder = config_.content_encoding_encoder; const auto content_encoding_decoders = config_.content_encoding_decoders; const auto header_optimization = BuildHeaderOptimizationPolicy(config_); - options.httpOptions.httpClientFactory_create_fn = [nodes, port, content_encoding_decoders, header_optimization] { - return NewAlternatorHttpClientFactory(nodes, port, content_encoding_decoders, header_optimization); + options.httpOptions.httpClientFactory_create_fn = [ + nodes, + port, + content_encoding_encoder, + content_encoding_decoders, + header_optimization] { + return NewAlternatorHttpClientFactory( + nodes, + port, + content_encoding_encoder, + content_encoding_decoders, + header_optimization); }; } diff --git a/src/config.cpp b/src/config.cpp index 8cd31a2..d523e9c 100644 --- a/src/config.cpp +++ b/src/config.cpp @@ -47,6 +47,7 @@ void ValidateConfig(const Config& config) { if (config.max_connections == 0) { throw std::invalid_argument("max_connections must be > 0"); } + (void)detail::BuildContentEncodingValue(config.content_encoding_encoder); (void)detail::BuildAcceptEncodingValue(config.content_encoding_decoders); if (config.key_route_affinity.partition_key_discovery_attempts == 0) { throw std::invalid_argument("partition_key_discovery_attempts must be > 0"); diff --git a/src/http_compression.cpp b/src/http_compression.cpp index 2b051a2..7553898 100644 --- a/src/http_compression.cpp +++ b/src/http_compression.cpp @@ -42,11 +42,54 @@ struct ContentEncodingDecoderEntry { #if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB [[nodiscard]] uInt CheckedZlibSize(std::size_t size) { if (size > std::numeric_limits::max()) { - throw std::runtime_error("compressed HTTP response is too large"); + throw std::runtime_error("HTTP content is too large for zlib"); } return static_cast(size); } +[[nodiscard]] std::string DeflateBody(const std::string& body, int window_bits) { + z_stream stream{}; + const auto init_code = deflateInit2( + &stream, + Z_DEFAULT_COMPRESSION, + Z_DEFLATED, + window_bits, + 8, + Z_DEFAULT_STRATEGY); + if (init_code != Z_OK) { + throw std::runtime_error("deflateInit2 failed"); + } + + struct DeflateGuard { + z_stream* stream; + ~DeflateGuard() { + deflateEnd(stream); + } + } guard{&stream}; + + auto* input = reinterpret_cast(body.data()); + stream.next_in = const_cast(input); + stream.avail_in = CheckedZlibSize(body.size()); + + std::array buffer{}; + std::string output; + while (true) { + stream.next_out = reinterpret_cast(buffer.data()); + stream.avail_out = static_cast(buffer.size()); + + const auto code = deflate(&stream, Z_FINISH); + const auto produced = buffer.size() - stream.avail_out; + output.append(buffer.data(), produced); + + if (code == Z_STREAM_END) { + return output; + } + if (code != Z_OK) { + throw std::runtime_error("failed to deflate HTTP request"); + } + } +} + [[nodiscard]] std::string InflateBody(const std::string& body, int window_bits) { z_stream stream{}; const auto init_code = inflateInit2(&stream, window_bits); @@ -117,6 +160,22 @@ struct ContentEncodingDecoderEntry { return encodings; } +[[nodiscard]] std::string NormalizeRequestEncoding( + const std::shared_ptr& content_encoding_encoder) { + if (!content_encoding_encoder) { + return {}; + } + + auto encoding = NormalizeResponseEncoding(content_encoding_encoder->ContentEncoding()); + if (encoding.empty()) { + throw std::invalid_argument("content_encoding_encoder must not advertise an empty encoding"); + } + if (encoding.find(',') != std::string::npos) { + throw std::invalid_argument("content_encoding_encoder must advertise exactly one encoding"); + } + return encoding; +} + [[nodiscard]] std::vector BuildDecoderEntries( const std::vector>& content_encoding_decoders) { std::vector entries; @@ -174,6 +233,11 @@ struct ContentEncodingDecoderEntry { } // namespace +std::string BuildContentEncodingValue( + const std::shared_ptr& content_encoding_encoder) { + return NormalizeRequestEncoding(content_encoding_encoder); +} + std::string ToLowerAscii(std::string value) { std::transform(value.begin(), value.end(), value.begin(), [](unsigned char ch) { return static_cast(std::tolower(ch)); @@ -231,6 +295,25 @@ std::string FindHttpHeaderValue(const std::string& headers, const std::string& n namespace scylladb::alternator { +GzipContentEncodingEncoder::GzipContentEncodingEncoder() { +#if !SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB + throw std::invalid_argument("zlib request encoding is not available"); +#endif +} + +std::string GzipContentEncodingEncoder::ContentEncoding() const { + return "gzip"; +} + +std::string GzipContentEncodingEncoder::Encode(std::string body) const { +#if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB + return detail::DeflateBody(body, MAX_WBITS + 16); +#else + (void)body; + throw std::runtime_error("zlib request encoding is not available"); +#endif +} + ZlibContentEncodingDecoder::ZlibContentEncodingDecoder(std::vector accepted_response_encodings) : accepted_response_encodings_(detail::NormalizeZlibResponseEncodings(std::move(accepted_response_encodings))) {} diff --git a/src/http_compression.h b/src/http_compression.h index 32e9c43..0f87fae 100644 --- a/src/http_compression.h +++ b/src/http_compression.h @@ -7,6 +7,8 @@ namespace scylladb::alternator::detail { +[[nodiscard]] std::string BuildContentEncodingValue( + const std::shared_ptr& content_encoding_encoder); [[nodiscard]] std::string BuildAcceptEncodingValue( const std::vector>& content_encoding_decoders); [[nodiscard]] std::string DecodeHttpResponseBody( diff --git a/tests/aws_dynamodb_helper_test.cpp b/tests/aws_dynamodb_helper_test.cpp index 6fc53e3..22e7584 100644 --- a/tests/aws_dynamodb_helper_test.cpp +++ b/tests/aws_dynamodb_helper_test.cpp @@ -374,6 +374,14 @@ std::map RequestHeaders(const std::string& request) { return headers; } +std::string RequestBody(const std::string& request) { + const auto header_end = request.find("\r\n\r\n"); + if (header_end == std::string::npos) { + return {}; + } + return request.substr(header_end + 4); +} + #if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB std::string CompressBody(const std::string& body, int window_bits) { if (body.size() > std::numeric_limits::max()) { @@ -422,6 +430,47 @@ std::string CompressBody(const std::string& body, int window_bits) { } } +std::string DecompressBody(const std::string& body, int window_bits) { + if (body.size() > std::numeric_limits::max()) { + throw std::runtime_error("body too large to decompress"); + } + + z_stream stream{}; + const auto init_code = inflateInit2(&stream, window_bits); + if (init_code != Z_OK) { + throw std::runtime_error("inflateInit2 failed"); + } + + struct InflateGuard { + z_stream* stream; + ~InflateGuard() { + inflateEnd(stream); + } + } guard{&stream}; + + auto* input = reinterpret_cast(body.data()); + stream.next_in = const_cast(input); + stream.avail_in = static_cast(body.size()); + + std::array buffer{}; + std::string output; + while (true) { + stream.next_out = reinterpret_cast(buffer.data()); + stream.avail_out = static_cast(buffer.size()); + + const auto code = inflate(&stream, Z_NO_FLUSH); + const auto produced = buffer.size() - stream.avail_out; + output.append(buffer.data(), produced); + + if (code == Z_STREAM_END) { + return output; + } + if (code != Z_OK) { + throw std::runtime_error("failed to inflate compressed HTTP body"); + } + } +} + SequencedHttpResponse GzipJsonResponse(const std::string& json) { SequencedHttpResponse response; response.body = CompressBody(json, MAX_WBITS + 16); @@ -1025,6 +1074,117 @@ TEST(AwsDynamoDBHelper, HttpClientFactoryDoesNotAdvertiseCompressionForNonAltern #endif } +TEST(AwsDynamoDBHelper, HttpClientFactoryCompressesGzipRequests) { +#if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB + KeepAliveSequenceHttpServer server({ + {200, "OK", "{}"}, + }); + + auto cfg = DiscoveryTestConfig(server.Port()); + cfg.content_encoding_encoder = std::make_shared(); + cfg.header_optimization = std::make_shared(std::vector{ + "Host", + "X-Amz-Target", + "Content-Length", + }); + + aws::DynamoDBHelper helper({"127.0.0.1"}, cfg); + + Aws::SDKOptions sdk_options; + helper.ApplyToSDKOptions(sdk_options); + AwsApiGuard api(sdk_options); + + auto client_config = helper.NewClientConfiguration(); + client_config.retryStrategy = Aws::MakeShared( + "AlternatorClientCppRequestCompressionTestRetryStrategy", + 0, + 0); + client_config.version = Aws::Http::Version::HTTP_VERSION_1_1; + + Aws::Auth::AWSCredentials credentials("alternator", "secret"); + Aws::DynamoDB::DynamoDBClient client(credentials, helper.NewEndpointProvider(), client_config); + + Aws::DynamoDB::Model::PutItemRequest request; + request.SetTableName("orders"); + request.AddItem("id", AwsStringValue("order-123")); + request.AddItem("payload", AwsStringValue("created")); + auto outcome = client.PutItem(request); + EXPECT_TRUE(outcome.IsSuccess()) << outcome.GetError().GetMessage(); + + server.Wait(); + + ASSERT_EQ(server.Requests().size(), 1U); + const auto headers = RequestHeaders(server.Requests()[0]); + const auto body = RequestBody(server.Requests()[0]); + + const auto content_encoding = headers.find("content-encoding"); + ASSERT_NE(content_encoding, headers.end()); + EXPECT_EQ(content_encoding->second, "gzip"); + + const auto content_length = headers.find("content-length"); + ASSERT_NE(content_length, headers.end()); + EXPECT_EQ(content_length->second, std::to_string(body.size())); + EXPECT_EQ(headers.find("x-amz-content-sha256"), headers.end()); + + const auto decoded_body = DecompressBody(body, MAX_WBITS + 16); + EXPECT_NE(decoded_body.find(R"("TableName":"orders")"), std::string::npos); + EXPECT_NE(decoded_body.find(R"("S":"order-123")"), std::string::npos); +#else + GTEST_SKIP() << "zlib support is not enabled"; +#endif +} + +TEST(AwsDynamoDBHelper, HttpClientFactoryDoesNotCompressRequestsForNonAlternatorEndpoints) { +#if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB + KeepAliveSequenceHttpServer server({ + {200, "OK", "{}"}, + }); + + Config cfg; + cfg.port = server.Port(); + cfg.scheme = "http"; + cfg.aws_region = "us-east-1"; + cfg.credentials = {"alternator", "secret"}; + cfg.nodes_list_update_period = std::chrono::milliseconds{0}; + cfg.http_client_timeout = std::chrono::milliseconds{2000}; + cfg.connect_timeout = std::chrono::milliseconds{1000}; + cfg.content_encoding_encoder = std::make_shared(); + + aws::DynamoDBHelper helper({"node1.example.com"}, cfg); + + Aws::SDKOptions sdk_options; + helper.ApplyToSDKOptions(sdk_options); + AwsApiGuard api(sdk_options); + + auto client_config = helper.NewClientConfiguration(); + client_config.endpointOverride = Aws::String(("http://127.0.0.1:" + std::to_string(server.Port())).c_str()); + client_config.retryStrategy = Aws::MakeShared( + "AlternatorClientCppRequestCompressionNonAlternatorTestRetryStrategy", + 0, + 0); + client_config.version = Aws::Http::Version::HTTP_VERSION_1_1; + + Aws::Auth::AWSCredentials credentials("alternator", "secret"); + Aws::DynamoDB::DynamoDBClient client(credentials, client_config); + + Aws::DynamoDB::Model::PutItemRequest request; + request.SetTableName("orders"); + request.AddItem("id", AwsStringValue("order-123")); + request.AddItem("payload", AwsStringValue("created")); + auto outcome = client.PutItem(request); + EXPECT_TRUE(outcome.IsSuccess()) << outcome.GetError().GetMessage(); + + server.Wait(); + + ASSERT_EQ(server.Requests().size(), 1U); + const auto headers = RequestHeaders(server.Requests()[0]); + EXPECT_EQ(headers.find("content-encoding"), headers.end()); + EXPECT_NE(RequestBody(server.Requests()[0]).find(R"("TableName":"orders")"), std::string::npos); +#else + GTEST_SKIP() << "zlib support is not enabled"; +#endif +} + TEST(AwsDynamoDBHelper, HandlesRepeatedNonSuccessDynamoDbResponses) { KeepAliveSequenceHttpServer server({ {400, "Bad Request", R"({"__type":"ValidationException","message":"bad"})"}, diff --git a/tests/http_client_test.cpp b/tests/http_client_test.cpp index 49999ec..91e5e2b 100644 --- a/tests/http_client_test.cpp +++ b/tests/http_client_test.cpp @@ -350,6 +350,19 @@ TEST(HttpClient, RequestsAndDecodesGzipResponse) { #endif } +TEST(HttpClient, GzipContentEncodingEncoderRoundTripsWithZlibDecoder) { +#if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB + const std::string body = R"({"TableName":"orders","Limit":10})"; + const GzipContentEncodingEncoder encoder; + const ZlibContentEncodingDecoder decoder({"gzip"}); + + EXPECT_EQ(encoder.ContentEncoding(), "gzip"); + EXPECT_EQ(decoder.Decode(encoder.Encode(body), encoder.ContentEncoding()), body); +#else + GTEST_SKIP() << "zlib support is not enabled"; +#endif +} + TEST(HttpClient, CanUseExplicitZlibContentEncodingDecoder) { #if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB const std::string body = "[\"node1.local\"]"; From e61c302ff4c8ed719cbf12454b3b2dbfce9369ea Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Mon, 13 Jul 2026 18:14:08 -0400 Subject: [PATCH 2/7] DRIVER-409 Rename request compression API --- README.md | 12 +++---- include/scylladb/alternator/config.h | 14 ++++---- src/aws_dynamodb_helper.cpp | 50 ++++++++++++++-------------- src/config.cpp | 2 +- src/http_compression.cpp | 22 ++++++------ src/http_compression.h | 4 +-- tests/aws_dynamodb_helper_test.cpp | 4 +-- tests/http_client_test.cpp | 8 ++--- 8 files changed, 58 insertions(+), 58 deletions(-) diff --git a/README.md b/README.md index 6b73a6a..a0dec4d 100644 --- a/README.md +++ b/README.md @@ -142,7 +142,7 @@ before `Aws::InitAPI()`. ### HTTP Request Compression -Request compression is disabled by default. Configure a request content encoder +Request compression is disabled by default. Configure a request compressor before constructing `DynamoDBHelper`, then install the Alternator HTTP client factory with `DynamoDBHelper::ApplyToSDKOptions()` before `Aws::InitAPI()`. The factory compresses request bodies only for requests routed to known @@ -151,13 +151,13 @@ headers before sending the request. ```cpp scylladb::alternator::Config cfg; -cfg.content_encoding_encoder = - std::make_shared(); +cfg.request_compressor = + std::make_shared(); ``` -Only gzip request compression is built in. `GzipContentEncodingEncoder` requires -zlib at build time. The `HttpContentEncodingEncoder` interface keeps request -compression extensible for other content encodings. +Only gzip request compression is built in. `GzipRequestCompressor` requires +zlib at build time. The `HttpRequestCompressor` interface keeps request +compression extensible for other HTTP content encodings. ### HTTP Response Compression diff --git a/include/scylladb/alternator/config.h b/include/scylladb/alternator/config.h index 7795b10..4a0b105 100644 --- a/include/scylladb/alternator/config.h +++ b/include/scylladb/alternator/config.h @@ -25,12 +25,12 @@ class HttpContentEncodingDecoder { [[nodiscard]] virtual std::string Decode(std::string body, const std::string& content_encoding) const = 0; }; -class HttpContentEncodingEncoder { +class HttpRequestCompressor { public: - virtual ~HttpContentEncodingEncoder() = default; + virtual ~HttpRequestCompressor() = default; [[nodiscard]] virtual std::string ContentEncoding() const = 0; - [[nodiscard]] virtual std::string Encode(std::string body) const = 0; + [[nodiscard]] virtual std::string Compress(std::string body) const = 0; }; class ZlibContentEncodingDecoder final : public HttpContentEncodingDecoder { @@ -44,12 +44,12 @@ class ZlibContentEncodingDecoder final : public HttpContentEncodingDecoder { std::vector accepted_response_encodings_; }; -class GzipContentEncodingEncoder final : public HttpContentEncodingEncoder { +class GzipRequestCompressor final : public HttpRequestCompressor { public: - GzipContentEncodingEncoder(); + GzipRequestCompressor(); [[nodiscard]] std::string ContentEncoding() const override; - [[nodiscard]] std::string Encode(std::string body) const override; + [[nodiscard]] std::string Compress(std::string body) const override; }; struct HeaderOptimizationContext { @@ -103,7 +103,7 @@ struct Config { unsigned max_connections = 100; bool reuse_discovery_connections = true; - std::shared_ptr content_encoding_encoder; + std::shared_ptr request_compressor; std::vector> content_encoding_decoders; std::string user_agent = "scylladb-alternator-client-cpp/devel"; std::shared_ptr header_optimization; diff --git a/src/aws_dynamodb_helper.cpp b/src/aws_dynamodb_helper.cpp index a937739..d531752 100644 --- a/src/aws_dynamodb_helper.cpp +++ b/src/aws_dynamodb_helper.cpp @@ -155,20 +155,20 @@ std::shared_ptr NewRequestBodyStream(const std::string& value) { return stream; } -void EncodeAwsRequestBody( +void CompressAwsRequestBody( const std::shared_ptr& request, - const std::shared_ptr& content_encoding_encoder, - const std::string& content_encoding_value) { - if (!request || !content_encoding_encoder || content_encoding_value.empty() || + const std::shared_ptr& request_compressor, + const std::string& request_content_encoding) { + if (!request || !request_compressor || request_content_encoding.empty() || !request->GetContentBody() || request->IsEventStreamRequest() || request->HasTransferEncoding() || request->HasContentEncoding()) { return; } - const auto encoded_body = content_encoding_encoder->Encode(ReadRequestBody(*request->GetContentBody())); - request->AddContentBody(NewRequestBodyStream(encoded_body)); - request->SetContentEncoding(Aws::String(content_encoding_value.c_str())); - request->SetContentLength(Aws::String(std::to_string(encoded_body.size()).c_str())); + const auto compressed_body = request_compressor->Compress(ReadRequestBody(*request->GetContentBody())); + request->AddContentBody(NewRequestBodyStream(compressed_body)); + request->SetContentEncoding(Aws::String(request_content_encoding.c_str())); + request->SetContentLength(Aws::String(std::to_string(compressed_body.size()).c_str())); request->DeleteHeader("x-amz-content-sha256"); } @@ -249,7 +249,7 @@ HeaderOptimizationPolicy BuildHeaderOptimizationPolicy(const Config& config) { const auto headers = config.header_optimization->AllowedHeaders( HeaderOptimizationContextFromConfig(config)); - if (config.content_encoding_encoder) { + if (config.request_compressor) { auto with_compression_headers = headers; with_compression_headers.push_back(Aws::Http::CONTENT_ENCODING_HEADER); with_compression_headers.push_back(Aws::Http::CONTENT_LENGTH_HEADER); @@ -296,14 +296,14 @@ class AlternatorHttpClient final : public Aws::Http::HttpClient { public: AlternatorHttpClient(std::shared_ptr nodes, std::uint16_t endpoint_override_port, - std::shared_ptr content_encoding_encoder, + std::shared_ptr request_compressor, std::vector> content_encoding_decoders, HeaderOptimizationPolicy header_optimization, std::shared_ptr delegate) : nodes_(std::move(nodes)) , endpoint_override_port_(endpoint_override_port) - , content_encoding_encoder_(std::move(content_encoding_encoder)) - , content_encoding_value_(detail::BuildContentEncodingValue(content_encoding_encoder_)) + , request_compressor_(std::move(request_compressor)) + , request_content_encoding_(detail::BuildRequestContentEncodingValue(request_compressor_)) , content_encoding_decoders_(std::move(content_encoding_decoders)) , accept_encoding_value_(detail::BuildAcceptEncodingValue(content_encoding_decoders_)) , header_optimization_(std::move(header_optimization)) @@ -325,7 +325,7 @@ class AlternatorHttpClient final : public Aws::Http::HttpClient { request->SetAcceptEncoding(Aws::String(accept_encoding_value_.c_str())); } if (!node.Empty()) { - EncodeAwsRequestBody(request, content_encoding_encoder_, content_encoding_value_); + CompressAwsRequestBody(request, request_compressor_, request_content_encoding_); OptimizeRequestHeaders(request, header_optimization_); } @@ -349,8 +349,8 @@ class AlternatorHttpClient final : public Aws::Http::HttpClient { private: std::shared_ptr nodes_; std::uint16_t endpoint_override_port_ = 0; - std::shared_ptr content_encoding_encoder_; - std::string content_encoding_value_; + std::shared_ptr request_compressor_; + std::string request_content_encoding_; std::vector> content_encoding_decoders_; std::string accept_encoding_value_; HeaderOptimizationPolicy header_optimization_; @@ -361,12 +361,12 @@ class AlternatorHttpClientFactory final : public Aws::Http::HttpClientFactory { public: AlternatorHttpClientFactory(std::shared_ptr nodes, std::uint16_t endpoint_override_port, - std::shared_ptr content_encoding_encoder, + std::shared_ptr request_compressor, std::vector> content_encoding_decoders, HeaderOptimizationPolicy header_optimization) : nodes_(std::move(nodes)) , endpoint_override_port_(endpoint_override_port) - , content_encoding_encoder_(std::move(content_encoding_encoder)) + , request_compressor_(std::move(request_compressor)) , content_encoding_decoders_(std::move(content_encoding_decoders)) , header_optimization_(std::move(header_optimization)) { if (!nodes_) { @@ -385,7 +385,7 @@ class AlternatorHttpClientFactory final : public Aws::Http::HttpClientFactory { kAllocationTag, nodes_, endpoint_override_port_, - content_encoding_encoder_, + request_compressor_, content_encoding_decoders_, header_optimization_, std::move(delegate)); @@ -421,7 +421,7 @@ class AlternatorHttpClientFactory final : public Aws::Http::HttpClientFactory { private: std::shared_ptr nodes_; std::uint16_t endpoint_override_port_ = 0; - std::shared_ptr content_encoding_encoder_; + std::shared_ptr request_compressor_; std::vector> content_encoding_decoders_; HeaderOptimizationPolicy header_optimization_; }; @@ -433,14 +433,14 @@ std::string EndpointOverride(std::uint16_t port, const std::string& scheme) { std::shared_ptr NewAlternatorHttpClientFactory( std::shared_ptr nodes, std::uint16_t endpoint_override_port, - std::shared_ptr content_encoding_encoder, + std::shared_ptr request_compressor, std::vector> content_encoding_decoders, HeaderOptimizationPolicy header_optimization) { return Aws::MakeShared( kAllocationTag, std::move(nodes), endpoint_override_port, - std::move(content_encoding_encoder), + std::move(request_compressor), std::move(content_encoding_decoders), std::move(header_optimization)); } @@ -854,7 +854,7 @@ std::shared_ptr DynamoDBHelper::NewHttpClientFacto return NewAlternatorHttpClientFactory( nodes_, config_.port, - config_.content_encoding_encoder, + config_.request_compressor, config_.content_encoding_decoders, BuildHeaderOptimizationPolicy(config_)); } @@ -886,19 +886,19 @@ std::shared_ptr DynamoDBHelper::Nodes() const { void DynamoDBHelper::ApplyToSDKOptions(Aws::SDKOptions& options) const { auto nodes = nodes_; const auto port = config_.port; - const auto content_encoding_encoder = config_.content_encoding_encoder; + const auto request_compressor = config_.request_compressor; const auto content_encoding_decoders = config_.content_encoding_decoders; const auto header_optimization = BuildHeaderOptimizationPolicy(config_); options.httpOptions.httpClientFactory_create_fn = [ nodes, port, - content_encoding_encoder, + request_compressor, content_encoding_decoders, header_optimization] { return NewAlternatorHttpClientFactory( nodes, port, - content_encoding_encoder, + request_compressor, content_encoding_decoders, header_optimization); }; diff --git a/src/config.cpp b/src/config.cpp index d523e9c..f19c406 100644 --- a/src/config.cpp +++ b/src/config.cpp @@ -47,7 +47,7 @@ void ValidateConfig(const Config& config) { if (config.max_connections == 0) { throw std::invalid_argument("max_connections must be > 0"); } - (void)detail::BuildContentEncodingValue(config.content_encoding_encoder); + (void)detail::BuildRequestContentEncodingValue(config.request_compressor); (void)detail::BuildAcceptEncodingValue(config.content_encoding_decoders); if (config.key_route_affinity.partition_key_discovery_attempts == 0) { throw std::invalid_argument("partition_key_discovery_attempts must be > 0"); diff --git a/src/http_compression.cpp b/src/http_compression.cpp index 7553898..cfa18af 100644 --- a/src/http_compression.cpp +++ b/src/http_compression.cpp @@ -161,17 +161,17 @@ struct ContentEncodingDecoderEntry { } [[nodiscard]] std::string NormalizeRequestEncoding( - const std::shared_ptr& content_encoding_encoder) { - if (!content_encoding_encoder) { + const std::shared_ptr& request_compressor) { + if (!request_compressor) { return {}; } - auto encoding = NormalizeResponseEncoding(content_encoding_encoder->ContentEncoding()); + auto encoding = NormalizeResponseEncoding(request_compressor->ContentEncoding()); if (encoding.empty()) { - throw std::invalid_argument("content_encoding_encoder must not advertise an empty encoding"); + throw std::invalid_argument("request_compressor must not advertise an empty encoding"); } if (encoding.find(',') != std::string::npos) { - throw std::invalid_argument("content_encoding_encoder must advertise exactly one encoding"); + throw std::invalid_argument("request_compressor must advertise exactly one encoding"); } return encoding; } @@ -233,9 +233,9 @@ struct ContentEncodingDecoderEntry { } // namespace -std::string BuildContentEncodingValue( - const std::shared_ptr& content_encoding_encoder) { - return NormalizeRequestEncoding(content_encoding_encoder); +std::string BuildRequestContentEncodingValue( + const std::shared_ptr& request_compressor) { + return NormalizeRequestEncoding(request_compressor); } std::string ToLowerAscii(std::string value) { @@ -295,17 +295,17 @@ std::string FindHttpHeaderValue(const std::string& headers, const std::string& n namespace scylladb::alternator { -GzipContentEncodingEncoder::GzipContentEncodingEncoder() { +GzipRequestCompressor::GzipRequestCompressor() { #if !SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB throw std::invalid_argument("zlib request encoding is not available"); #endif } -std::string GzipContentEncodingEncoder::ContentEncoding() const { +std::string GzipRequestCompressor::ContentEncoding() const { return "gzip"; } -std::string GzipContentEncodingEncoder::Encode(std::string body) const { +std::string GzipRequestCompressor::Compress(std::string body) const { #if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB return detail::DeflateBody(body, MAX_WBITS + 16); #else diff --git a/src/http_compression.h b/src/http_compression.h index 0f87fae..fd14bdb 100644 --- a/src/http_compression.h +++ b/src/http_compression.h @@ -7,8 +7,8 @@ namespace scylladb::alternator::detail { -[[nodiscard]] std::string BuildContentEncodingValue( - const std::shared_ptr& content_encoding_encoder); +[[nodiscard]] std::string BuildRequestContentEncodingValue( + const std::shared_ptr& request_compressor); [[nodiscard]] std::string BuildAcceptEncodingValue( const std::vector>& content_encoding_decoders); [[nodiscard]] std::string DecodeHttpResponseBody( diff --git a/tests/aws_dynamodb_helper_test.cpp b/tests/aws_dynamodb_helper_test.cpp index 22e7584..4aa2718 100644 --- a/tests/aws_dynamodb_helper_test.cpp +++ b/tests/aws_dynamodb_helper_test.cpp @@ -1081,7 +1081,7 @@ TEST(AwsDynamoDBHelper, HttpClientFactoryCompressesGzipRequests) { }); auto cfg = DiscoveryTestConfig(server.Port()); - cfg.content_encoding_encoder = std::make_shared(); + cfg.request_compressor = std::make_shared(); cfg.header_optimization = std::make_shared(std::vector{ "Host", "X-Amz-Target", @@ -1148,7 +1148,7 @@ TEST(AwsDynamoDBHelper, HttpClientFactoryDoesNotCompressRequestsForNonAlternator cfg.nodes_list_update_period = std::chrono::milliseconds{0}; cfg.http_client_timeout = std::chrono::milliseconds{2000}; cfg.connect_timeout = std::chrono::milliseconds{1000}; - cfg.content_encoding_encoder = std::make_shared(); + cfg.request_compressor = std::make_shared(); aws::DynamoDBHelper helper({"node1.example.com"}, cfg); diff --git a/tests/http_client_test.cpp b/tests/http_client_test.cpp index 91e5e2b..754e339 100644 --- a/tests/http_client_test.cpp +++ b/tests/http_client_test.cpp @@ -350,14 +350,14 @@ TEST(HttpClient, RequestsAndDecodesGzipResponse) { #endif } -TEST(HttpClient, GzipContentEncodingEncoderRoundTripsWithZlibDecoder) { +TEST(HttpClient, GzipRequestCompressorRoundTripsWithZlibDecoder) { #if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB const std::string body = R"({"TableName":"orders","Limit":10})"; - const GzipContentEncodingEncoder encoder; + const GzipRequestCompressor compressor; const ZlibContentEncodingDecoder decoder({"gzip"}); - EXPECT_EQ(encoder.ContentEncoding(), "gzip"); - EXPECT_EQ(decoder.Decode(encoder.Encode(body), encoder.ContentEncoding()), body); + EXPECT_EQ(compressor.ContentEncoding(), "gzip"); + EXPECT_EQ(decoder.Decode(compressor.Compress(body), compressor.ContentEncoding()), body); #else GTEST_SKIP() << "zlib support is not enabled"; #endif From 0392a86256042130dee8e0089f62d04ae74cae28 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Mon, 13 Jul 2026 18:25:09 -0400 Subject: [PATCH 3/7] DRIVER-409 Stream request compression API --- README.md | 7 +- include/scylladb/alternator/config.h | 18 +++- src/aws_dynamodb_helper.cpp | 78 +++++++++++---- src/http_compression.cpp | 69 +++++++++---- tests/aws_dynamodb_helper_test.cpp | 142 +++++++++++++++++++++++++-- tests/aws_integration_test.cpp | 17 ++++ tests/http_client_test.cpp | 21 +++- 7 files changed, 296 insertions(+), 56 deletions(-) diff --git a/README.md b/README.md index a0dec4d..567972a 100644 --- a/README.md +++ b/README.md @@ -156,8 +156,11 @@ cfg.request_compressor = ``` Only gzip request compression is built in. `GzipRequestCompressor` requires -zlib at build time. The `HttpRequestCompressor` interface keeps request -compression extensible for other HTTP content encodings. +zlib at build time and skips bodies smaller than 1024 bytes by default. Pass a +different minimum size, such as `GzipRequestCompressor(0)`, to control that +policy. The `HttpRequestCompressor` interface owns the decision to compress and +compresses from an input stream to an output stream so implementations do not +need to copy the whole request body into an intermediate string. ### HTTP Response Compression diff --git a/include/scylladb/alternator/config.h b/include/scylladb/alternator/config.h index 4a0b105..8ced978 100644 --- a/include/scylladb/alternator/config.h +++ b/include/scylladb/alternator/config.h @@ -2,6 +2,7 @@ #include #include +#include #include #include #include @@ -30,7 +31,12 @@ class HttpRequestCompressor { virtual ~HttpRequestCompressor() = default; [[nodiscard]] virtual std::string ContentEncoding() const = 0; - [[nodiscard]] virtual std::string Compress(std::string body) const = 0; + // Returns false to leave the request uncompressed, for example when + // input_size is below an implementation-defined minimum. + [[nodiscard]] virtual bool Compress( + std::istream& input, + std::uint64_t input_size, + std::ostream& output) const = 0; }; class ZlibContentEncodingDecoder final : public HttpContentEncodingDecoder { @@ -46,10 +52,16 @@ class ZlibContentEncodingDecoder final : public HttpContentEncodingDecoder { class GzipRequestCompressor final : public HttpRequestCompressor { public: - GzipRequestCompressor(); + explicit GzipRequestCompressor(std::uint64_t min_size_bytes = 1024); [[nodiscard]] std::string ContentEncoding() const override; - [[nodiscard]] std::string Compress(std::string body) const override; + [[nodiscard]] bool Compress( + std::istream& input, + std::uint64_t input_size, + std::ostream& output) const override; + +private: + std::uint64_t min_size_bytes_ = 0; }; struct HeaderOptimizationContext { diff --git a/src/aws_dynamodb_helper.cpp b/src/aws_dynamodb_helper.cpp index d531752..c6d84f0 100644 --- a/src/aws_dynamodb_helper.cpp +++ b/src/aws_dynamodb_helper.cpp @@ -126,15 +126,6 @@ std::string ReadResponseBody(Aws::IOStream& body) { return std::string(std::istreambuf_iterator(body), std::istreambuf_iterator()); } -std::string ReadRequestBody(Aws::IOStream& body) { - body.clear(); - body.seekg(0, std::ios::beg); - auto value = std::string(std::istreambuf_iterator(body), std::istreambuf_iterator()); - body.clear(); - body.seekg(0, std::ios::beg); - return value; -} - void WriteResponseBody(Aws::IOStream& body, const std::string& value) { if (!value.empty()) { body.write(value.data(), static_cast(value.size())); @@ -144,17 +135,45 @@ void WriteResponseBody(Aws::IOStream& body, const std::string& value) { body.seekg(0, std::ios::beg); } -std::shared_ptr NewRequestBodyStream(const std::string& value) { +std::shared_ptr NewRequestBodyStream() { auto stream = Aws::MakeShared(kAllocationTag); - if (!value.empty()) { - stream->write(value.data(), static_cast(value.size())); - } - stream->flush(); - stream->clear(); - stream->seekg(0, std::ios::beg); return stream; } +std::size_t PrepareOriginalRequestBodyForReading(Aws::IOStream& body) { + body.clear(); + body.seekg(0, std::ios::end); + if (!body) { + throw std::runtime_error("HTTP request body is not seekable"); + } + + const auto size = body.tellg(); + if (size == std::streampos(-1)) { + throw std::runtime_error("HTTP request body size is unavailable"); + } + + body.clear(); + body.seekg(0, std::ios::beg); + if (!body) { + throw std::runtime_error("HTTP request body is not seekable"); + } + return static_cast(size); +} + +std::size_t PrepareCompressedRequestBodyForReading(Aws::IOStream& body) { + body.flush(); + const auto size = body.tellp(); + if (size == std::streampos(-1)) { + throw std::runtime_error("compressed HTTP request body size is unavailable"); + } + body.clear(); + body.seekg(0, std::ios::beg); + if (!body) { + throw std::runtime_error("compressed HTTP request body is not seekable"); + } + return static_cast(size); +} + void CompressAwsRequestBody( const std::shared_ptr& request, const std::shared_ptr& request_compressor, @@ -165,10 +184,26 @@ void CompressAwsRequestBody( return; } - const auto compressed_body = request_compressor->Compress(ReadRequestBody(*request->GetContentBody())); - request->AddContentBody(NewRequestBodyStream(compressed_body)); + auto& original_body = *request->GetContentBody(); + const auto original_body_size = PrepareOriginalRequestBodyForReading(original_body); + + auto compressed_body = NewRequestBodyStream(); + const auto compressed = request_compressor->Compress( + original_body, + original_body_size, + *compressed_body); + if (!compressed) { + original_body.clear(); + original_body.seekg(0, std::ios::beg); + if (!original_body) { + throw std::runtime_error("HTTP request body is not seekable"); + } + return; + } + const auto compressed_body_size = PrepareCompressedRequestBodyForReading(*compressed_body); request->SetContentEncoding(Aws::String(request_content_encoding.c_str())); - request->SetContentLength(Aws::String(std::to_string(compressed_body.size()).c_str())); + request->SetContentLength(Aws::String(std::to_string(compressed_body_size).c_str())); + request->AddContentBody(std::move(compressed_body)); request->DeleteHeader("x-amz-content-sha256"); } @@ -325,7 +360,10 @@ class AlternatorHttpClient final : public Aws::Http::HttpClient { request->SetAcceptEncoding(Aws::String(accept_encoding_value_.c_str())); } if (!node.Empty()) { - CompressAwsRequestBody(request, request_compressor_, request_content_encoding_); + CompressAwsRequestBody( + request, + request_compressor_, + request_content_encoding_); OptimizeRequestHeaders(request, header_optimization_); } diff --git a/src/http_compression.cpp b/src/http_compression.cpp index cfa18af..3e4900c 100644 --- a/src/http_compression.cpp +++ b/src/http_compression.cpp @@ -12,6 +12,8 @@ #if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB #include #endif +#include +#include #include #include #include @@ -47,7 +49,7 @@ struct ContentEncodingDecoderEntry { return static_cast(size); } -[[nodiscard]] std::string DeflateBody(const std::string& body, int window_bits) { +void DeflateBody(std::istream& input, std::ostream& output, int window_bits) { z_stream stream{}; const auto init_code = deflateInit2( &stream, @@ -67,25 +69,42 @@ struct ContentEncodingDecoderEntry { } } guard{&stream}; - auto* input = reinterpret_cast(body.data()); - stream.next_in = const_cast(input); - stream.avail_in = CheckedZlibSize(body.size()); - + std::array input_buffer{}; std::array buffer{}; - std::string output; while (true) { - stream.next_out = reinterpret_cast(buffer.data()); - stream.avail_out = static_cast(buffer.size()); + input.read(input_buffer.data(), static_cast(input_buffer.size())); + const auto read_size = input.gcount(); + if (input.bad()) { + throw std::runtime_error("failed to read HTTP request body"); + } - const auto code = deflate(&stream, Z_FINISH); - const auto produced = buffer.size() - stream.avail_out; - output.append(buffer.data(), produced); + stream.next_in = reinterpret_cast(input_buffer.data()); + stream.avail_in = CheckedZlibSize(static_cast(read_size)); + const auto flush = input.eof() ? Z_FINISH : Z_NO_FLUSH; + + do { + stream.next_out = reinterpret_cast(buffer.data()); + stream.avail_out = static_cast(buffer.size()); + + const auto code = deflate(&stream, flush); + const auto produced = buffer.size() - stream.avail_out; + if (produced > 0) { + output.write(buffer.data(), static_cast(produced)); + if (!output) { + throw std::runtime_error("failed to write compressed HTTP request body"); + } + } - if (code == Z_STREAM_END) { - return output; - } - if (code != Z_OK) { - throw std::runtime_error("failed to deflate HTTP request"); + if (code == Z_STREAM_END) { + return; + } + if (code != Z_OK) { + throw std::runtime_error("failed to deflate HTTP request"); + } + } while (stream.avail_in != 0 || stream.avail_out == 0); + + if (flush == Z_FINISH) { + throw std::runtime_error("failed to finish deflating HTTP request"); } } } @@ -295,7 +314,8 @@ std::string FindHttpHeaderValue(const std::string& headers, const std::string& n namespace scylladb::alternator { -GzipRequestCompressor::GzipRequestCompressor() { +GzipRequestCompressor::GzipRequestCompressor(std::uint64_t min_size_bytes) + : min_size_bytes_(min_size_bytes) { #if !SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB throw std::invalid_argument("zlib request encoding is not available"); #endif @@ -305,11 +325,20 @@ std::string GzipRequestCompressor::ContentEncoding() const { return "gzip"; } -std::string GzipRequestCompressor::Compress(std::string body) const { +bool GzipRequestCompressor::Compress( + std::istream& input, + std::uint64_t input_size, + std::ostream& output) const { #if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB - return detail::DeflateBody(body, MAX_WBITS + 16); + if (input_size < min_size_bytes_) { + return false; + } + detail::DeflateBody(input, output, MAX_WBITS + 16); + return true; #else - (void)body; + (void)input; + (void)input_size; + (void)output; throw std::runtime_error("zlib request encoding is not available"); #endif } diff --git a/tests/aws_dynamodb_helper_test.cpp b/tests/aws_dynamodb_helper_test.cpp index 4aa2718..504ab3d 100644 --- a/tests/aws_dynamodb_helper_test.cpp +++ b/tests/aws_dynamodb_helper_test.cpp @@ -523,6 +523,35 @@ Aws::DynamoDB::Model::AttributeValue AwsStringValue(const std::string& value) { return out; } +Aws::DynamoDB::Model::PutItemRequest NewPutItemRequest(std::string payload = "created") { + Aws::DynamoDB::Model::PutItemRequest request; + request.SetTableName("orders"); + request.AddItem("id", AwsStringValue("order-123")); + request.AddItem("payload", AwsStringValue(payload)); + return request; +} + +class DecliningRequestCompressor final : public HttpRequestCompressor { +public: + [[nodiscard]] std::string ContentEncoding() const override { + return "gzip"; + } + + [[nodiscard]] bool Compress( + std::istream& input, + std::uint64_t, + std::ostream& output) const override { + char buffer[256]; + while (input.read(buffer, static_cast(sizeof(buffer))) || input.gcount() > 0) { + } + if (input.bad()) { + throw std::runtime_error("failed to read request body"); + } + output << "ignored"; + return false; + } +}; + std::string PartitionKeyForHost(const aws::DynamoDBHelper& helper, const std::string& table_name, const std::string& host) { @@ -1104,10 +1133,7 @@ TEST(AwsDynamoDBHelper, HttpClientFactoryCompressesGzipRequests) { Aws::Auth::AWSCredentials credentials("alternator", "secret"); Aws::DynamoDB::DynamoDBClient client(credentials, helper.NewEndpointProvider(), client_config); - Aws::DynamoDB::Model::PutItemRequest request; - request.SetTableName("orders"); - request.AddItem("id", AwsStringValue("order-123")); - request.AddItem("payload", AwsStringValue("created")); + auto request = NewPutItemRequest(std::string(2048, 'x')); auto outcome = client.PutItem(request); EXPECT_TRUE(outcome.IsSuccess()) << outcome.GetError().GetMessage(); @@ -1124,16 +1150,117 @@ TEST(AwsDynamoDBHelper, HttpClientFactoryCompressesGzipRequests) { const auto content_length = headers.find("content-length"); ASSERT_NE(content_length, headers.end()); EXPECT_EQ(content_length->second, std::to_string(body.size())); + EXPECT_EQ(headers.find("transfer-encoding"), headers.end()); EXPECT_EQ(headers.find("x-amz-content-sha256"), headers.end()); + EXPECT_EQ(headers.find("content-type"), headers.end()); const auto decoded_body = DecompressBody(body, MAX_WBITS + 16); EXPECT_NE(decoded_body.find(R"("TableName":"orders")"), std::string::npos); EXPECT_NE(decoded_body.find(R"("S":"order-123")"), std::string::npos); + EXPECT_NE(decoded_body.find(std::string(2048, 'x')), std::string::npos); #else GTEST_SKIP() << "zlib support is not enabled"; #endif } +TEST(AwsDynamoDBHelper, HttpClientFactoryDoesNotCompressRequestsBelowMinimumSize) { +#if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB + KeepAliveSequenceHttpServer server({ + {200, "OK", "{}"}, + }); + + auto cfg = DiscoveryTestConfig(server.Port()); + cfg.request_compressor = std::make_shared(); + cfg.header_optimization = std::make_shared(std::vector{ + "Host", + "X-Amz-Target", + "Content-Length", + }); + + aws::DynamoDBHelper helper({"127.0.0.1"}, cfg); + + Aws::SDKOptions sdk_options; + helper.ApplyToSDKOptions(sdk_options); + AwsApiGuard api(sdk_options); + + auto client_config = helper.NewClientConfiguration(); + client_config.retryStrategy = Aws::MakeShared( + "AlternatorClientCppRequestCompressionMinimumSizeTestRetryStrategy", + 0, + 0); + client_config.version = Aws::Http::Version::HTTP_VERSION_1_1; + + Aws::Auth::AWSCredentials credentials("alternator", "secret"); + Aws::DynamoDB::DynamoDBClient client(credentials, helper.NewEndpointProvider(), client_config); + + auto request = NewPutItemRequest(); + auto outcome = client.PutItem(request); + EXPECT_TRUE(outcome.IsSuccess()) << outcome.GetError().GetMessage(); + + server.Wait(); + + ASSERT_EQ(server.Requests().size(), 1U); + const auto headers = RequestHeaders(server.Requests()[0]); + const auto body = RequestBody(server.Requests()[0]); + EXPECT_EQ(headers.find("content-encoding"), headers.end()); + const auto content_length = headers.find("content-length"); + ASSERT_NE(content_length, headers.end()); + EXPECT_EQ(content_length->second, std::to_string(body.size())); + EXPECT_EQ(headers.find("transfer-encoding"), headers.end()); + EXPECT_EQ(headers.find("content-type"), headers.end()); + EXPECT_NE(body.find(R"("TableName":"orders")"), std::string::npos); +#else + GTEST_SKIP() << "zlib support is not enabled"; +#endif +} + +TEST(AwsDynamoDBHelper, HttpClientFactoryKeepsOriginalRequestWhenCompressorDeclinesAfterReading) { + KeepAliveSequenceHttpServer server({ + {200, "OK", "{}"}, + }); + + auto cfg = DiscoveryTestConfig(server.Port()); + cfg.request_compressor = std::make_shared(); + cfg.header_optimization = std::make_shared(std::vector{ + "Host", + "X-Amz-Target", + "Content-Length", + }); + + aws::DynamoDBHelper helper({"127.0.0.1"}, cfg); + + Aws::SDKOptions sdk_options; + helper.ApplyToSDKOptions(sdk_options); + AwsApiGuard api(sdk_options); + + auto client_config = helper.NewClientConfiguration(); + client_config.retryStrategy = Aws::MakeShared( + "AlternatorClientCppRequestCompressionDeclinedTestRetryStrategy", + 0, + 0); + client_config.version = Aws::Http::Version::HTTP_VERSION_1_1; + + Aws::Auth::AWSCredentials credentials("alternator", "secret"); + Aws::DynamoDB::DynamoDBClient client(credentials, helper.NewEndpointProvider(), client_config); + + auto request = NewPutItemRequest(); + auto outcome = client.PutItem(request); + EXPECT_TRUE(outcome.IsSuccess()) << outcome.GetError().GetMessage(); + + server.Wait(); + + ASSERT_EQ(server.Requests().size(), 1U); + const auto headers = RequestHeaders(server.Requests()[0]); + const auto body = RequestBody(server.Requests()[0]); + EXPECT_EQ(headers.find("content-encoding"), headers.end()); + const auto content_length = headers.find("content-length"); + ASSERT_NE(content_length, headers.end()); + EXPECT_EQ(content_length->second, std::to_string(body.size())); + EXPECT_EQ(headers.find("transfer-encoding"), headers.end()); + EXPECT_EQ(body.find("ignored"), std::string::npos); + EXPECT_NE(body.find(R"("TableName":"orders")"), std::string::npos); +} + TEST(AwsDynamoDBHelper, HttpClientFactoryDoesNotCompressRequestsForNonAlternatorEndpoints) { #if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB KeepAliveSequenceHttpServer server({ @@ -1148,7 +1275,7 @@ TEST(AwsDynamoDBHelper, HttpClientFactoryDoesNotCompressRequestsForNonAlternator cfg.nodes_list_update_period = std::chrono::milliseconds{0}; cfg.http_client_timeout = std::chrono::milliseconds{2000}; cfg.connect_timeout = std::chrono::milliseconds{1000}; - cfg.request_compressor = std::make_shared(); + cfg.request_compressor = std::make_shared(0); aws::DynamoDBHelper helper({"node1.example.com"}, cfg); @@ -1167,10 +1294,7 @@ TEST(AwsDynamoDBHelper, HttpClientFactoryDoesNotCompressRequestsForNonAlternator Aws::Auth::AWSCredentials credentials("alternator", "secret"); Aws::DynamoDB::DynamoDBClient client(credentials, client_config); - Aws::DynamoDB::Model::PutItemRequest request; - request.SetTableName("orders"); - request.AddItem("id", AwsStringValue("order-123")); - request.AddItem("payload", AwsStringValue("created")); + auto request = NewPutItemRequest(); auto outcome = client.PutItem(request); EXPECT_TRUE(outcome.IsSuccess()) << outcome.GetError().GetMessage(); diff --git a/tests/aws_integration_test.cpp b/tests/aws_integration_test.cpp index aed78cf..35d26b4 100644 --- a/tests/aws_integration_test.cpp +++ b/tests/aws_integration_test.cpp @@ -14,6 +14,7 @@ #include #include +#include #include #include #include @@ -160,6 +161,22 @@ TEST(AwsDynamoDBIntegration, DynamoDBOperationsHttp) { RunDynamoDBOperations(IntegrationConfig(IntegrationHttpPort()), "cpp_integration_http"); } +TEST(AwsDynamoDBIntegration, DynamoDBOperationsHttpWithRequestCompressionAndHeaderOptimization) { + REQUIRE_INTEGRATION(); +#if !SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB + GTEST_SKIP() << "zlib support is not enabled"; +#endif + + auto cfg = IntegrationConfig(IntegrationHttpPort()); + cfg.request_compressor = std::make_shared(0); + cfg.header_optimization = std::make_shared(std::vector{ + "Host", + "X-Amz-Target", + "Content-Length", + }); + RunDynamoDBOperations(std::move(cfg), "cpp_integration_http_gzip_request"); +} + TEST(AwsDynamoDBIntegration, DynamoDBOperationsHttpsWithoutCertificateVerification) { REQUIRE_INTEGRATION(); diff --git a/tests/http_client_test.cpp b/tests/http_client_test.cpp index 754e339..c0c6116 100644 --- a/tests/http_client_test.cpp +++ b/tests/http_client_test.cpp @@ -353,11 +353,28 @@ TEST(HttpClient, RequestsAndDecodesGzipResponse) { TEST(HttpClient, GzipRequestCompressorRoundTripsWithZlibDecoder) { #if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB const std::string body = R"({"TableName":"orders","Limit":10})"; - const GzipRequestCompressor compressor; + const GzipRequestCompressor compressor(0); const ZlibContentEncodingDecoder decoder({"gzip"}); + std::istringstream input(body); + std::ostringstream output; EXPECT_EQ(compressor.ContentEncoding(), "gzip"); - EXPECT_EQ(decoder.Decode(compressor.Compress(body), compressor.ContentEncoding()), body); + EXPECT_TRUE(compressor.Compress(input, body.size(), output)); + EXPECT_EQ(decoder.Decode(output.str(), compressor.ContentEncoding()), body); +#else + GTEST_SKIP() << "zlib support is not enabled"; +#endif +} + +TEST(HttpClient, GzipRequestCompressorSkipsBodiesBelowMinimumSize) { +#if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB + const std::string body = R"({"TableName":"orders"})"; + const GzipRequestCompressor compressor(1024); + std::istringstream input(body); + std::ostringstream output; + + EXPECT_FALSE(compressor.Compress(input, body.size(), output)); + EXPECT_EQ(output.str(), ""); #else GTEST_SKIP() << "zlib support is not enabled"; #endif From 0ff9b96bf08d151b1a623adf83b9b8330e32fd01 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Mon, 13 Jul 2026 18:26:13 -0400 Subject: [PATCH 4/7] DRIVER-409 Guard request compression stream reads --- src/http_compression.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/http_compression.cpp b/src/http_compression.cpp index 3e4900c..dae1f3c 100644 --- a/src/http_compression.cpp +++ b/src/http_compression.cpp @@ -74,7 +74,7 @@ void DeflateBody(std::istream& input, std::ostream& output, int window_bits) { while (true) { input.read(input_buffer.data(), static_cast(input_buffer.size())); const auto read_size = input.gcount(); - if (input.bad()) { + if (input.bad() || (input.fail() && !input.eof())) { throw std::runtime_error("failed to read HTTP request body"); } From 92a3a6ec5d7ae73f6003e4b3e0bb5dfc5eeebe2d Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Thu, 16 Jul 2026 13:47:59 -0400 Subject: [PATCH 5/7] DRIVER-409 Parse DynamoDB request bodies in tests --- tests/aws_dynamodb_helper_test.cpp | 37 +++++++++++++++++++++++++----- 1 file changed, 31 insertions(+), 6 deletions(-) diff --git a/tests/aws_dynamodb_helper_test.cpp b/tests/aws_dynamodb_helper_test.cpp index 504ab3d..492fc94 100644 --- a/tests/aws_dynamodb_helper_test.cpp +++ b/tests/aws_dynamodb_helper_test.cpp @@ -2,6 +2,7 @@ #include #include +#include #include #include #include @@ -382,6 +383,32 @@ std::string RequestBody(const std::string& request) { return request.substr(header_end + 4); } +std::string ToStdString(const Aws::String& value) { + return std::string(value.data(), value.size()); +} + +void ExpectPutItemRequestBody(const std::string& body, const std::string& payload = "created") { + Aws::Utils::Json::JsonValue json(Aws::String(body.data(), body.size())); + ASSERT_TRUE(json.WasParseSuccessful()) << json.GetErrorMessage() << "\n" << body; + + const auto root = json.View(); + ASSERT_TRUE(root.ValueExists("TableName")) << body; + EXPECT_EQ(ToStdString(root.GetString("TableName")), "orders"); + + ASSERT_TRUE(root.ValueExists("Item")) << body; + const auto item = root.GetObject("Item"); + + ASSERT_TRUE(item.ValueExists("id")) << body; + const auto id = item.GetObject("id"); + ASSERT_TRUE(id.ValueExists("S")) << body; + EXPECT_EQ(ToStdString(id.GetString("S")), "order-123"); + + ASSERT_TRUE(item.ValueExists("payload")) << body; + const auto payload_item = item.GetObject("payload"); + ASSERT_TRUE(payload_item.ValueExists("S")) << body; + EXPECT_EQ(ToStdString(payload_item.GetString("S")), payload); +} + #if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB std::string CompressBody(const std::string& body, int window_bits) { if (body.size() > std::numeric_limits::max()) { @@ -1155,9 +1182,7 @@ TEST(AwsDynamoDBHelper, HttpClientFactoryCompressesGzipRequests) { EXPECT_EQ(headers.find("content-type"), headers.end()); const auto decoded_body = DecompressBody(body, MAX_WBITS + 16); - EXPECT_NE(decoded_body.find(R"("TableName":"orders")"), std::string::npos); - EXPECT_NE(decoded_body.find(R"("S":"order-123")"), std::string::npos); - EXPECT_NE(decoded_body.find(std::string(2048, 'x')), std::string::npos); + ExpectPutItemRequestBody(decoded_body, std::string(2048, 'x')); #else GTEST_SKIP() << "zlib support is not enabled"; #endif @@ -1208,7 +1233,7 @@ TEST(AwsDynamoDBHelper, HttpClientFactoryDoesNotCompressRequestsBelowMinimumSize EXPECT_EQ(content_length->second, std::to_string(body.size())); EXPECT_EQ(headers.find("transfer-encoding"), headers.end()); EXPECT_EQ(headers.find("content-type"), headers.end()); - EXPECT_NE(body.find(R"("TableName":"orders")"), std::string::npos); + ExpectPutItemRequestBody(body); #else GTEST_SKIP() << "zlib support is not enabled"; #endif @@ -1258,7 +1283,7 @@ TEST(AwsDynamoDBHelper, HttpClientFactoryKeepsOriginalRequestWhenCompressorDecli EXPECT_EQ(content_length->second, std::to_string(body.size())); EXPECT_EQ(headers.find("transfer-encoding"), headers.end()); EXPECT_EQ(body.find("ignored"), std::string::npos); - EXPECT_NE(body.find(R"("TableName":"orders")"), std::string::npos); + ExpectPutItemRequestBody(body); } TEST(AwsDynamoDBHelper, HttpClientFactoryDoesNotCompressRequestsForNonAlternatorEndpoints) { @@ -1303,7 +1328,7 @@ TEST(AwsDynamoDBHelper, HttpClientFactoryDoesNotCompressRequestsForNonAlternator ASSERT_EQ(server.Requests().size(), 1U); const auto headers = RequestHeaders(server.Requests()[0]); EXPECT_EQ(headers.find("content-encoding"), headers.end()); - EXPECT_NE(RequestBody(server.Requests()[0]).find(R"("TableName":"orders")"), std::string::npos); + ExpectPutItemRequestBody(RequestBody(server.Requests()[0])); #else GTEST_SKIP() << "zlib support is not enabled"; #endif From c750eeb60a2c7723db039801fb5dfd029f7d0c43 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Thu, 16 Jul 2026 14:27:49 -0400 Subject: [PATCH 6/7] DRIVER-409 Document Alternator SigV4 limits --- README.md | 26 +++++++++++++++++++++----- 1 file changed, 21 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 567972a..40179b3 100644 --- a/README.md +++ b/README.md @@ -71,10 +71,12 @@ int main() { ``` `Config::aws_region` is required by the AWS SDK for C++ client configuration, -including request signing and diagnostic metadata. Alternator does not use that -value to choose endpoints; `DynamoDBHelper` discovers nodes through -`/localnodes` and installs the endpoint provider or HTTP wrapper that routes -requests to live Alternator nodes. The default value is +including request signing and diagnostic metadata. Alternator does not support +SigV4 request validation, and this client does not try to preserve a valid +SigV4 signature after Alternator-specific request rewriting. Alternator does +not use the region value to choose endpoints; `DynamoDBHelper` discovers nodes +through `/localnodes` and installs the endpoint provider or HTTP wrapper that +routes requests to live Alternator nodes. The default value is `default-alb-region`, which prevents the SDK configuration from being empty but can look confusing in logs, traces, or metrics. @@ -85,7 +87,16 @@ Alternator node discovery or load balancing. The AWS adapter uses the SDK's per-client `DynamoDBEndpointProviderBase` hook by default. AWS SDK for C++ resolves that endpoint once for a retried operation, so `DynamoDBHelper::ApplyToSDKOptions()` can also install a process-wide HTTP client factory before `Aws::InitAPI()`. That factory delegates to the SDK HTTP client and rotates only requests already aimed at the helper's Alternator endpoints. -The HTTP client factory runs after request signing. This is the closest public AWS SDK for C++ hook for retry-aware endpoint rewriting and header optimization, but it means applications that depend on strict SigV4 host validation should prefer endpoint-provider routing or implement their own SDK wrapper. Automatic request-content-based endpoint rewriting is not available in the same form. The core library does provide deterministic key-route affinity primitives, and `DynamoDBHelper` exposes them for applications that integrate request-specific routing in their own AWS SDK wrapper. +The HTTP client factory runs after request signing. This is the closest public +AWS SDK for C++ hook for retry-aware endpoint rewriting, header optimization, +and request or response compression, and it matches Alternator's SigV4 behavior: +Alternator does not validate SigV4 signatures. Do not use the factory through +a proxy, gateway, or deployment policy that requires valid SigV4 signatures +after the request leaves the SDK. Automatic request-content-based endpoint +rewriting is not available in the same form. The core library does provide +deterministic key-route affinity primitives, and `DynamoDBHelper` exposes them +for applications that integrate request-specific routing in their own AWS SDK +wrapper. ### Headers Optimization @@ -149,6 +160,11 @@ The factory compresses request bodies only for requests routed to known Alternator nodes and sets the matching `Content-Encoding` and `Content-Length` headers before sending the request. +Request compression is applied by the Alternator HTTP client factory after the +AWS SDK signs the request. That is compatible with Alternator because Alternator +does not support SigV4 validation. It is not compatible with any proxy, gateway, +or DynamoDB-compatible service that validates SigV4 signatures. + ```cpp scylladb::alternator::Config cfg; cfg.request_compressor = From ca9bdb3f982b44269e8ce30f422a5caf193b1512 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Thu, 16 Jul 2026 14:48:13 -0400 Subject: [PATCH 7/7] DRIVER-409 Simplify request compression helpers --- README.md | 9 +++++---- src/aws_dynamodb_helper.cpp | 37 +++++++++++++------------------------ 2 files changed, 18 insertions(+), 28 deletions(-) diff --git a/README.md b/README.md index 40179b3..f85b46d 100644 --- a/README.md +++ b/README.md @@ -127,10 +127,11 @@ cfg.header_optimization = }); ``` -Header names are matched case-insensitively. The override is exact; if it is -set, the helper does not add credential or user-agent headers automatically. -Applications can provide their own implementation by deriving from -`HeaderOptimization`: +Header names are matched case-insensitively. The override is exact except that +request compression always keeps `Content-Encoding` and `Content-Length` when +`Config::request_compressor` is set. If the override is set, the helper does +not add credential or user-agent headers automatically. Applications can provide +their own implementation by deriving from `HeaderOptimization`: ```cpp class MyHeaderOptimization final : public scylladb::alternator::HeaderOptimization { diff --git a/src/aws_dynamodb_helper.cpp b/src/aws_dynamodb_helper.cpp index c6d84f0..91a8be0 100644 --- a/src/aws_dynamodb_helper.cpp +++ b/src/aws_dynamodb_helper.cpp @@ -135,9 +135,12 @@ void WriteResponseBody(Aws::IOStream& body, const std::string& value) { body.seekg(0, std::ios::beg); } -std::shared_ptr NewRequestBodyStream() { - auto stream = Aws::MakeShared(kAllocationTag); - return stream; +void RewindForReading(Aws::IOStream& body, const std::string& description) { + body.clear(); + body.seekg(0, std::ios::beg); + if (!body) { + throw std::runtime_error(description + " is not seekable"); + } } std::size_t PrepareOriginalRequestBodyForReading(Aws::IOStream& body) { @@ -152,11 +155,7 @@ std::size_t PrepareOriginalRequestBodyForReading(Aws::IOStream& body) { throw std::runtime_error("HTTP request body size is unavailable"); } - body.clear(); - body.seekg(0, std::ios::beg); - if (!body) { - throw std::runtime_error("HTTP request body is not seekable"); - } + RewindForReading(body, "HTTP request body"); return static_cast(size); } @@ -166,11 +165,7 @@ std::size_t PrepareCompressedRequestBodyForReading(Aws::IOStream& body) { if (size == std::streampos(-1)) { throw std::runtime_error("compressed HTTP request body size is unavailable"); } - body.clear(); - body.seekg(0, std::ios::beg); - if (!body) { - throw std::runtime_error("compressed HTTP request body is not seekable"); - } + RewindForReading(body, "compressed HTTP request body"); return static_cast(size); } @@ -187,17 +182,13 @@ void CompressAwsRequestBody( auto& original_body = *request->GetContentBody(); const auto original_body_size = PrepareOriginalRequestBodyForReading(original_body); - auto compressed_body = NewRequestBodyStream(); + auto compressed_body = Aws::MakeShared(kAllocationTag); const auto compressed = request_compressor->Compress( original_body, original_body_size, *compressed_body); if (!compressed) { - original_body.clear(); - original_body.seekg(0, std::ios::beg); - if (!original_body) { - throw std::runtime_error("HTTP request body is not seekable"); - } + RewindForReading(original_body, "HTTP request body"); return; } const auto compressed_body_size = PrepareCompressedRequestBodyForReading(*compressed_body); @@ -282,13 +273,11 @@ HeaderOptimizationPolicy BuildHeaderOptimizationPolicy(const Config& config) { return {}; } - const auto headers = config.header_optimization->AllowedHeaders( + auto headers = config.header_optimization->AllowedHeaders( HeaderOptimizationContextFromConfig(config)); if (config.request_compressor) { - auto with_compression_headers = headers; - with_compression_headers.push_back(Aws::Http::CONTENT_ENCODING_HEADER); - with_compression_headers.push_back(Aws::Http::CONTENT_LENGTH_HEADER); - return {true, NormalizeHeaderAllowlist(with_compression_headers)}; + headers.push_back(Aws::Http::CONTENT_ENCODING_HEADER); + headers.push_back(Aws::Http::CONTENT_LENGTH_HEADER); } return {true, NormalizeHeaderAllowlist(headers)}; }