DRIVER-409 Add gzip request compression

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)

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<scylladb::alternator::GzipRequestCompressor>();
+```
+
+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.

include/scylladb/alternator/config.h

@@ -2,6 +2,7 @@
 
 #include <chrono>
 #include <cstdint>
+#include <iosfwd>
 #include <memory>
 #include <string>
 #include <vector>
@@ -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<std::string> accepted_response_encodings = {"gzip", "deflate"});
@@ -36,6 +50,20 @@ class ZlibContentEncodingDecoder final : public HttpContentEncodingDecoder {
     std::vector<std::string> 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<HttpRequestCompressor> request_compressor;
     std::vector<std::shared_ptr<HttpContentEncodingDecoder>> content_encoding_decoders;
     std::string user_agent = "scylladb-alternator-client-cpp/devel";
     std::shared_ptr<HeaderOptimization> header_optimization;

src/aws_dynamodb_helper.cpp

@@ -13,6 +13,7 @@
 #include <aws/core/http/standard/StandardHttpRequest.h>
 #include <aws/core/http/standard/StandardHttpResponse.h>
 #include <aws/core/utils/logging/LogMacros.h>
+#include <aws/core/utils/memory/stl/AWSStringStream.h>
 #include <aws/dynamodb/model/AttributeAction.h>
 #include <aws/dynamodb/model/AttributeValueUpdate.h>
 #include <aws/dynamodb/model/BatchWriteItemRequest.h>
@@ -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<std::size_t>(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<std::size_t>(size);
+}
+
+void CompressAwsRequestBody(
+    const std::shared_ptr<Aws::Http::HttpRequest>& request,
+    const std::shared_ptr<HttpRequestCompressor>& 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<Aws::StringStream>(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<Aws::Http::HttpResponse> DecodeCompressedAwsResponse(
     const std::shared_ptr<Aws::Http::HttpRequest>& request,
     const std::shared_ptr<Aws::Http::HttpResponse>& 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<AlternatorLiveNodes> nodes,
                          std::uint16_t endpoint_override_port,
+                         std::shared_ptr<HttpRequestCompressor> request_compressor,
                          std::vector<std::shared_ptr<HttpContentEncodingDecoder>> content_encoding_decoders,
                          HeaderOptimizationPolicy header_optimization,
                          std::shared_ptr<Aws::Http::HttpClient> 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<AlternatorLiveNodes> nodes_;
     std::uint16_t endpoint_override_port_ = 0;
+    std::shared_ptr<HttpRequestCompressor> request_compressor_;
+    std::string request_content_encoding_;
     std::vector<std::shared_ptr<HttpContentEncodingDecoder>> 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<AlternatorLiveNodes> nodes,
                                 std::uint16_t endpoint_override_port,
+                                std::shared_ptr<HttpRequestCompressor> request_compressor,
                                 std::vector<std::shared_ptr<HttpContentEncodingDecoder>> 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<AlternatorLiveNodes> nodes_;
     std::uint16_t endpoint_override_port_ = 0;
+    std::shared_ptr<HttpRequestCompressor> request_compressor_;
     std::vector<std::shared_ptr<HttpContentEncodingDecoder>> content_encoding_decoders_;
     HeaderOptimizationPolicy header_optimization_;
 };
@@ -379,12 +460,14 @@ std::string EndpointOverride(std::uint16_t port, const std::string& scheme) {
 std::shared_ptr<Aws::Http::HttpClientFactory> NewAlternatorHttpClientFactory(
     std::shared_ptr<AlternatorLiveNodes> nodes,
     std::uint16_t endpoint_override_port,
+    std::shared_ptr<HttpRequestCompressor> request_compressor,
     std::vector<std::shared_ptr<HttpContentEncodingDecoder>> content_encoding_decoders,
     HeaderOptimizationPolicy header_optimization) {
     return Aws::MakeShared<AlternatorHttpClientFactory>(
         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<Aws::Http::HttpClientFactory> DynamoDBHelper::NewHttpClientFacto
     return NewAlternatorHttpClientFactory(
         nodes_,
         config_.port,
+        config_.request_compressor,
         config_.content_encoding_decoders,
         BuildHeaderOptimizationPolicy(config_));
 }
@@ -829,10 +913,21 @@ std::shared_ptr<AlternatorLiveNodes> 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);
     };
 }
 

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");