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..f85b46d 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 @@ -116,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 { @@ -140,6 +152,33 @@ 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 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 +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 = + std::make_shared(); +``` + +Only gzip request compression is built in. `GzipRequestCompressor` requires +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 Response compression is disabled by default. When configured, the built-in @@ -378,6 +417,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..8ced978 100644 --- a/include/scylladb/alternator/config.h +++ b/include/scylladb/alternator/config.h @@ -2,6 +2,7 @@ #include #include +#include #include #include #include @@ -25,6 +26,19 @@ class HttpContentEncodingDecoder { [[nodiscard]] virtual std::string Decode(std::string body, const std::string& content_encoding) const = 0; }; +class HttpRequestCompressor { +public: + virtual ~HttpRequestCompressor() = default; + + [[nodiscard]] virtual std::string ContentEncoding() 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 { public: explicit ZlibContentEncodingDecoder(std::vector accepted_response_encodings = {"gzip", "deflate"}); @@ -36,6 +50,20 @@ class ZlibContentEncodingDecoder final : public HttpContentEncodingDecoder { std::vector accepted_response_encodings_; }; +class GzipRequestCompressor final : public HttpRequestCompressor { +public: + explicit GzipRequestCompressor(std::uint64_t min_size_bytes = 1024); + + [[nodiscard]] std::string ContentEncoding() 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 { bool credentials_configured = false; bool user_agent_configured = false; @@ -87,6 +115,7 @@ struct Config { unsigned max_connections = 100; bool reuse_discovery_connections = true; + 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 3c8c777..91a8be0 100644 --- a/src/aws_dynamodb_helper.cpp +++ b/src/aws_dynamodb_helper.cpp @@ -13,6 +13,7 @@ #include #include #include +#include #include #include #include @@ -134,6 +135,69 @@ void WriteResponseBody(Aws::IOStream& body, const std::string& value) { body.seekg(0, std::ios::beg); } +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) { + 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"); + } + + RewindForReading(body, "HTTP request body"); + 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"); + } + RewindForReading(body, "compressed HTTP request body"); + return static_cast(size); +} + +void CompressAwsRequestBody( + const std::shared_ptr& request, + 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; + } + + auto& original_body = *request->GetContentBody(); + const auto original_body_size = PrepareOriginalRequestBodyForReading(original_body); + + auto compressed_body = Aws::MakeShared(kAllocationTag); + const auto compressed = request_compressor->Compress( + original_body, + original_body_size, + *compressed_body); + if (!compressed) { + RewindForReading(original_body, "HTTP request body"); + 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->AddContentBody(std::move(compressed_body)); + request->DeleteHeader("x-amz-content-sha256"); +} + std::shared_ptr DecodeCompressedAwsResponse( const std::shared_ptr& request, const std::shared_ptr& response, @@ -209,8 +273,12 @@ 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) { + headers.push_back(Aws::Http::CONTENT_ENCODING_HEADER); + headers.push_back(Aws::Http::CONTENT_LENGTH_HEADER); + } return {true, NormalizeHeaderAllowlist(headers)}; } @@ -252,11 +320,14 @@ class AlternatorHttpClient final : public Aws::Http::HttpClient { public: AlternatorHttpClient(std::shared_ptr nodes, std::uint16_t endpoint_override_port, + 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) + , 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)) @@ -278,6 +349,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_); OptimizeRequestHeaders(request, header_optimization_); } @@ -301,6 +376,8 @@ class AlternatorHttpClient final : public Aws::Http::HttpClient { private: std::shared_ptr nodes_; std::uint16_t endpoint_override_port_ = 0; + std::shared_ptr request_compressor_; + std::string request_content_encoding_; std::vector> content_encoding_decoders_; std::string accept_encoding_value_; HeaderOptimizationPolicy header_optimization_; @@ -311,10 +388,12 @@ class AlternatorHttpClientFactory final : public Aws::Http::HttpClientFactory { public: AlternatorHttpClientFactory(std::shared_ptr nodes, std::uint16_t endpoint_override_port, + std::shared_ptr request_compressor, std::vector> content_encoding_decoders, HeaderOptimizationPolicy header_optimization) : nodes_(std::move(nodes)) , endpoint_override_port_(endpoint_override_port) + , request_compressor_(std::move(request_compressor)) , content_encoding_decoders_(std::move(content_encoding_decoders)) , header_optimization_(std::move(header_optimization)) { if (!nodes_) { @@ -333,6 +412,7 @@ class AlternatorHttpClientFactory final : public Aws::Http::HttpClientFactory { kAllocationTag, nodes_, endpoint_override_port_, + request_compressor_, content_encoding_decoders_, header_optimization_, std::move(delegate)); @@ -368,6 +448,7 @@ class AlternatorHttpClientFactory final : public Aws::Http::HttpClientFactory { private: std::shared_ptr nodes_; std::uint16_t endpoint_override_port_ = 0; + std::shared_ptr request_compressor_; std::vector> content_encoding_decoders_; HeaderOptimizationPolicy header_optimization_; }; @@ -379,12 +460,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 request_compressor, std::vector> content_encoding_decoders, HeaderOptimizationPolicy header_optimization) { return Aws::MakeShared( kAllocationTag, std::move(nodes), endpoint_override_port, + std::move(request_compressor), std::move(content_encoding_decoders), std::move(header_optimization)); } @@ -798,6 +881,7 @@ std::shared_ptr DynamoDBHelper::NewHttpClientFacto return NewAlternatorHttpClientFactory( nodes_, config_.port, + config_.request_compressor, config_.content_encoding_decoders, BuildHeaderOptimizationPolicy(config_)); } @@ -829,10 +913,21 @@ std::shared_ptr DynamoDBHelper::Nodes() const { void DynamoDBHelper::ApplyToSDKOptions(Aws::SDKOptions& options) const { auto nodes = nodes_; const auto port = config_.port; + 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_decoders, header_optimization] { - return NewAlternatorHttpClientFactory(nodes, port, content_encoding_decoders, header_optimization); + options.httpOptions.httpClientFactory_create_fn = [ + nodes, + port, + request_compressor, + content_encoding_decoders, + header_optimization] { + return NewAlternatorHttpClientFactory( + nodes, + port, + request_compressor, + content_encoding_decoders, + header_optimization); }; } diff --git a/src/config.cpp b/src/config.cpp index 8cd31a2..f19c406 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::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 2b051a2..dae1f3c 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 @@ -42,11 +44,71 @@ 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); } +void DeflateBody(std::istream& input, std::ostream& output, 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}; + + std::array input_buffer{}; + std::array buffer{}; + while (true) { + input.read(input_buffer.data(), static_cast(input_buffer.size())); + const auto read_size = input.gcount(); + if (input.bad() || (input.fail() && !input.eof())) { + throw std::runtime_error("failed to read HTTP request body"); + } + + 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; + } + 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"); + } + } +} + [[nodiscard]] std::string InflateBody(const std::string& body, int window_bits) { z_stream stream{}; const auto init_code = inflateInit2(&stream, window_bits); @@ -117,6 +179,22 @@ struct ContentEncodingDecoderEntry { return encodings; } +[[nodiscard]] std::string NormalizeRequestEncoding( + const std::shared_ptr& request_compressor) { + if (!request_compressor) { + return {}; + } + + auto encoding = NormalizeResponseEncoding(request_compressor->ContentEncoding()); + if (encoding.empty()) { + throw std::invalid_argument("request_compressor must not advertise an empty encoding"); + } + if (encoding.find(',') != std::string::npos) { + throw std::invalid_argument("request_compressor must advertise exactly one encoding"); + } + return encoding; +} + [[nodiscard]] std::vector BuildDecoderEntries( const std::vector>& content_encoding_decoders) { std::vector entries; @@ -174,6 +252,11 @@ struct ContentEncodingDecoderEntry { } // namespace +std::string BuildRequestContentEncodingValue( + const std::shared_ptr& request_compressor) { + return NormalizeRequestEncoding(request_compressor); +} + 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 +314,35 @@ std::string FindHttpHeaderValue(const std::string& headers, const std::string& n namespace scylladb::alternator { +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 +} + +std::string GzipRequestCompressor::ContentEncoding() const { + return "gzip"; +} + +bool GzipRequestCompressor::Compress( + std::istream& input, + std::uint64_t input_size, + std::ostream& output) const { +#if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB + if (input_size < min_size_bytes_) { + return false; + } + detail::DeflateBody(input, output, MAX_WBITS + 16); + return true; +#else + (void)input; + (void)input_size; + (void)output; + 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..fd14bdb 100644 --- a/src/http_compression.h +++ b/src/http_compression.h @@ -7,6 +7,8 @@ namespace scylladb::alternator::detail { +[[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 6fc53e3..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 @@ -374,6 +375,40 @@ 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); +} + +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()) { @@ -422,6 +457,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); @@ -474,6 +550,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) { @@ -1025,6 +1130,210 @@ TEST(AwsDynamoDBHelper, HttpClientFactoryDoesNotAdvertiseCompressionForNonAltern #endif } +TEST(AwsDynamoDBHelper, HttpClientFactoryCompressesGzipRequests) { +#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( + "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); + + auto request = NewPutItemRequest(std::string(2048, 'x')); + 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("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); + ExpectPutItemRequestBody(decoded_body, std::string(2048, 'x')); +#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()); + ExpectPutItemRequestBody(body); +#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); + ExpectPutItemRequestBody(body); +} + +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.request_compressor = std::make_shared(0); + + 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); + + 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]); + EXPECT_EQ(headers.find("content-encoding"), headers.end()); + ExpectPutItemRequestBody(RequestBody(server.Requests()[0])); +#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/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 49999ec..c0c6116 100644 --- a/tests/http_client_test.cpp +++ b/tests/http_client_test.cpp @@ -350,6 +350,36 @@ TEST(HttpClient, RequestsAndDecodesGzipResponse) { #endif } +TEST(HttpClient, GzipRequestCompressorRoundTripsWithZlibDecoder) { +#if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB + const std::string body = R"({"TableName":"orders","Limit":10})"; + const GzipRequestCompressor compressor(0); + const ZlibContentEncodingDecoder decoder({"gzip"}); + std::istringstream input(body); + std::ostringstream output; + + EXPECT_EQ(compressor.ContentEncoding(), "gzip"); + 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 +} + TEST(HttpClient, CanUseExplicitZlibContentEncodingDecoder) { #if SCYLLADB_ALTERNATOR_CLIENT_CPP_HAS_ZLIB const std::string body = "[\"node1.local\"]";