Add replication compression setting

.github/workflows/ci.yml

@@ -76,7 +76,7 @@ jobs:
             exclude_api_version_tag: ""
             audit_enabled: "false"
           - reductstore_version: "latest"
-            exclude_api_version_tag: "~[1_20]"
+            exclude_api_version_tag: "~[1_21]"
             audit_enabled: "false"
 
     steps:

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Add replication compression setting support, [PR-137](https://github.com/reductstore/reduct-cpp/pull/137)
 - Add replication destination prefix support, [PR-135](https://github.com/reductstore/reduct-cpp/pull/135)
 
 ## 1.20.0 - 2026-06-16

src/reduct/client.h

@@ -203,6 +203,7 @@ class IClient {
    * Replication information
    */
   enum class ReplicationMode { kEnabled, kPaused, kDisabled };
+  enum class ReplicationCompression { kNone, kZstd, kGzip };
 
   struct ReplicationInfo {
     std::string name;                                  // Replication name
@@ -227,6 +228,8 @@ class IClient {
         entries;  // Entries to replicate. If empty, all entries are replicated. Wildcards are supported.
     std::optional<std::string> when;                   // Replication condition
     ReplicationMode mode = ReplicationMode::kEnabled;  // Replication mode
+    ReplicationCompression compression =
+        ReplicationCompression::kNone;  // Replication transfer compression
 
     auto operator<=>(const ReplicationSettings&) const = default;
   };

src/reduct/internal/serialisation.cc

@@ -31,6 +31,27 @@ IClient::ReplicationMode ParseReplicationMode(const nlohmann::json& mode_json) {
   throw std::invalid_argument("Invalid replication mode: " + mode);
 }
 
+IClient::ReplicationCompression ParseReplicationCompression(const nlohmann::json& compression_json) {
+  if (compression_json.is_null()) {
+    return IClient::ReplicationCompression::kNone;
+  }
+
+  const auto compression = compression_json.get<std::string>();
+  if (compression == "none") {
+    return IClient::ReplicationCompression::kNone;
+  }
+
+  if (compression == "zstd") {
+    return IClient::ReplicationCompression::kZstd;
+  }
+
+  if (compression == "gzip") {
+    return IClient::ReplicationCompression::kGzip;
+  }
+
+  throw std::invalid_argument("Invalid replication compression: " + compression);
+}
+
 IClient::LifecycleMode ParseLifecycleMode(const nlohmann::json& mode_json) {
   if (mode_json.is_null()) {
     return IClient::LifecycleMode::kEnabled;
@@ -83,6 +104,19 @@ std::string ReplicationModeToString(IClient::ReplicationMode mode) {
   throw std::invalid_argument("Invalid replication mode");
 }
 
+std::string ReplicationCompressionToString(IClient::ReplicationCompression compression) {
+  switch (compression) {
+    case IClient::ReplicationCompression::kNone:
+      return "none";
+    case IClient::ReplicationCompression::kZstd:
+      return "zstd";
+    case IClient::ReplicationCompression::kGzip:
+      return "gzip";
+  }
+
+  throw std::invalid_argument("Invalid replication compression");
+}
+
 std::string LifecycleModeToString(IClient::LifecycleMode mode) {
   switch (mode) {
     case IClient::LifecycleMode::kEnabled:
@@ -242,6 +276,9 @@ Result<nlohmann::json> ReplicationSettingsToJsonString(IClient::ReplicationSetti
     }
     json_data["entries"] = settings.entries;
     json_data["mode"] = ReplicationModeToString(settings.mode);
+    if (settings.compression != IClient::ReplicationCompression::kNone) {
+      json_data["compression"] = ReplicationCompressionToString(settings.compression);
+    }
 
     if (settings.when) {
       try {
@@ -277,6 +314,8 @@ Result<IClient::FullReplicationInfo> ParseFullReplicationInfo(const nlohmann::js
         .entries = settings.at("entries"),
         .mode =
             settings.contains("mode") ? ParseReplicationMode(settings.at("mode")) : IClient::ReplicationMode::kEnabled,
+        .compression = settings.contains("compression") ? ParseReplicationCompression(settings.at("compression"))
+                                                        : IClient::ReplicationCompression::kNone,
     };
 
     if (settings.contains("dst_token") && !settings.at("dst_token").is_null()) {

src/reduct/internal/serialisation.h

@@ -47,6 +47,13 @@ Result<std::vector<IClient::ReplicationInfo>> ParseReplicationList(const nlohman
  */
 std::string ReplicationModeToString(IClient::ReplicationMode mode);
 
+/**
+ * @brief Convert replication compression to string representation
+ * @param compression replication compression
+ * @return string value expected by API
+ */
+std::string ReplicationCompressionToString(IClient::ReplicationCompression compression);
+
 /**
  * @brief Serialize replication settings
  * @param settings to serialize

tests/reduct/replication_api_test.cc

@@ -23,6 +23,24 @@ IClient::ReplicationSettings DefaultSettings() {
   };
 }
 
+nlohmann::json DefaultReplicationResponse() {
+  return nlohmann::json{
+      {"info",
+       {{"name", "test_replication"},
+        {"mode", "enabled"},
+        {"is_active", true},
+        {"is_provisioned", false},
+        {"pending_records", 0}}},
+      {"settings",
+       {{"src_bucket", "test_bucket_1"},
+        {"dst_bucket", "test_bucket_2"},
+        {"dst_host", "http://127.0.0.1:8383"},
+        {"entries", {"entry-1"}},
+        {"mode", "enabled"}}},
+      {"diagnostics", {{"hourly", {{"ok", 0}, {"errored", 0}, {"errors", nlohmann::json::object()}}}}},
+  };
+}
+
 }  // namespace
 
 TEST_CASE("reduct::Client should get list of replications", "[replication_api][1_8]") {
@@ -100,21 +118,7 @@ TEST_CASE("reduct::Client should serialize replication destination prefix", "[re
 }
 
 TEST_CASE("reduct::Client should parse replication destination prefix", "[replication_api]") {
-  auto response = nlohmann::json{
-      {"info",
-       {{"name", "test_replication"},
-        {"mode", "enabled"},
-        {"is_active", true},
-        {"is_provisioned", false},
-        {"pending_records", 0}}},
-      {"settings",
-       {{"src_bucket", "test_bucket_1"},
-        {"dst_bucket", "test_bucket_2"},
-        {"dst_host", "http://127.0.0.1:8383"},
-        {"entries", {"entry-1"}},
-        {"mode", "enabled"}}},
-      {"diagnostics", {{"hourly", {{"ok", 0}, {"errored", 0}, {"errors", nlohmann::json::object()}}}}},
-  };
+  auto response = DefaultReplicationResponse();
 
   SECTION("keeps destination prefix empty when absent") {
     auto [replication, err] = reduct::internal::ParseFullReplicationInfo(response);
@@ -133,7 +137,85 @@ TEST_CASE("reduct::Client should parse replication destination prefix", "[replic
   }
 }
 
-TEST_CASE("reduct::Client should set replication destination prefix", "[replication_api][1_20]") {
+TEST_CASE("reduct::Client should serialize replication compression", "[replication_api]") {
+  auto settings = DefaultSettings();
+
+  SECTION("omits default compression") {
+    auto [json, err] = reduct::internal::ReplicationSettingsToJsonString(settings);
+
+    REQUIRE(err == Error::kOk);
+    REQUIRE_FALSE(json.contains("compression"));
+    REQUIRE(reduct::internal::ReplicationCompressionToString(settings.compression) == "none");
+  }
+
+  SECTION("serializes zstd compression") {
+    settings.compression = IClient::ReplicationCompression::kZstd;
+
+    auto [json, err] = reduct::internal::ReplicationSettingsToJsonString(settings);
+
+    REQUIRE(err == Error::kOk);
+    REQUIRE(json.at("compression") == "zstd");
+    REQUIRE(reduct::internal::ReplicationCompressionToString(settings.compression) == "zstd");
+  }
+
+  SECTION("serializes gzip compression") {
+    settings.compression = IClient::ReplicationCompression::kGzip;
+
+    auto [json, err] = reduct::internal::ReplicationSettingsToJsonString(settings);
+
+    REQUIRE(err == Error::kOk);
+    REQUIRE(json.at("compression") == "gzip");
+    REQUIRE(reduct::internal::ReplicationCompressionToString(settings.compression) == "gzip");
+  }
+}
+
+TEST_CASE("reduct::Client should parse replication compression", "[replication_api]") {
+  auto response = DefaultReplicationResponse();
+
+  SECTION("defaults missing compression to none") {
+    auto [replication, err] = reduct::internal::ParseFullReplicationInfo(response);
+
+    REQUIRE(err == Error::kOk);
+    REQUIRE(replication.settings.compression == IClient::ReplicationCompression::kNone);
+  }
+
+  SECTION("defaults null compression to none") {
+    response["settings"]["compression"] = nullptr;
+
+    auto [replication, err] = reduct::internal::ParseFullReplicationInfo(response);
+
+    REQUIRE(err == Error::kOk);
+    REQUIRE(replication.settings.compression == IClient::ReplicationCompression::kNone);
+  }
+
+  SECTION("parses zstd compression") {
+    response["settings"]["compression"] = "zstd";
+
+    auto [replication, err] = reduct::internal::ParseFullReplicationInfo(response);
+
+    REQUIRE(err == Error::kOk);
+    REQUIRE(replication.settings.compression == IClient::ReplicationCompression::kZstd);
+  }
+
+  SECTION("parses gzip compression") {
+    response["settings"]["compression"] = "gzip";
+
+    auto [replication, err] = reduct::internal::ParseFullReplicationInfo(response);
+
+    REQUIRE(err == Error::kOk);
+    REQUIRE(replication.settings.compression == IClient::ReplicationCompression::kGzip);
+  }
+
+  SECTION("rejects invalid compression") {
+    response["settings"]["compression"] = "snappy";
+
+    auto [replication, err] = reduct::internal::ParseFullReplicationInfo(response);
+
+    REQUIRE(err == Error{.code = -1, .message = "Invalid replication compression: snappy"});
+  }
+}
+
+TEST_CASE("reduct::Client should set replication destination prefix", "[replication_api][1_21]") {
   Fixture ctx;
   auto settings = DefaultSettings();
   settings.dst_prefix = "robot-1";
@@ -154,6 +236,27 @@ TEST_CASE("reduct::Client should set replication destination prefix", "[replicat
   REQUIRE(updated_replication.settings.dst_prefix == "line-a");
 }
 
+TEST_CASE("reduct::Client should set replication compression", "[replication_api][1_21]") {
+  Fixture ctx;
+  auto settings = DefaultSettings();
+  settings.compression = IClient::ReplicationCompression::kZstd;
+
+  auto err = ctx.client->CreateReplication("test_replication", settings);
+  REQUIRE(err == Error::kOk);
+
+  auto [replication, err_2] = ctx.client->GetReplication("test_replication");
+  REQUIRE(err_2 == Error::kOk);
+  REQUIRE(replication.settings.compression == IClient::ReplicationCompression::kZstd);
+
+  settings.compression = IClient::ReplicationCompression::kGzip;
+  err = ctx.client->UpdateReplication("test_replication", settings);
+  REQUIRE(err == Error::kOk);
+
+  auto [updated_replication, err_3] = ctx.client->GetReplication("test_replication");
+  REQUIRE(err_3 == Error::kOk);
+  REQUIRE(updated_replication.settings.compression == IClient::ReplicationCompression::kGzip);
+}
+
 TEST_CASE("reduct::Client should set replication mode", "[replication_api][1_18]") {
   Fixture ctx;
   auto settings = DefaultSettings();