From d051c0ed416837ccfb28df36a4110a488a5011a7 Mon Sep 17 00:00:00 2001 From: Mother 6000 Date: Tue, 14 Jul 2026 12:19:44 +0000 Subject: [PATCH 1/3] Add replication compression setting --- .github/workflows/ci.yml | 2 +- src/reduct/client.h | 3 + src/reduct/internal/serialisation.cc | 39 ++++++++ src/reduct/internal/serialisation.h | 7 ++ tests/reduct/replication_api_test.cc | 133 ++++++++++++++++++++++++--- 5 files changed, 168 insertions(+), 16 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5fac00f..fdfdf36 100644 --- a/.github/workflows/ci.yml +++ b/.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_20]~[1_21]" audit_enabled: "false" steps: diff --git a/src/reduct/client.h b/src/reduct/client.h index afa306c..e612b63 100644 --- a/src/reduct/client.h +++ b/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 when; // Replication condition ReplicationMode mode = ReplicationMode::kEnabled; // Replication mode + ReplicationCompression compression = + ReplicationCompression::kNone; // Replication transfer compression auto operator<=>(const ReplicationSettings&) const = default; }; diff --git a/src/reduct/internal/serialisation.cc b/src/reduct/internal/serialisation.cc index 300486d..704bc2a 100644 --- a/src/reduct/internal/serialisation.cc +++ b/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(); + 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 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 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()) { diff --git a/src/reduct/internal/serialisation.h b/src/reduct/internal/serialisation.h index b935142..1ed8bcd 100644 --- a/src/reduct/internal/serialisation.h +++ b/src/reduct/internal/serialisation.h @@ -47,6 +47,13 @@ Result> 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 diff --git a/tests/reduct/replication_api_test.cc b/tests/reduct/replication_api_test.cc index 9e98e78..0f4c720 100644 --- a/tests/reduct/replication_api_test.cc +++ b/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,6 +137,84 @@ TEST_CASE("reduct::Client should parse replication destination prefix", "[replic } } +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_20]") { Fixture ctx; auto settings = DefaultSettings(); @@ -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(); From 810d518ef5d1252b32640f7e7286f46a19345689 Mon Sep 17 00:00:00 2001 From: Mother 6000 Date: Tue, 14 Jul 2026 12:23:12 +0000 Subject: [PATCH 2/3] Update changelog for PR #137 --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 289553e..e192fb2 100644 --- a/CHANGELOG.md +++ b/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 From 80d5936a4f53d29d745a430410cc12c51798c49b Mon Sep 17 00:00:00 2001 From: Mother 6000 Date: Tue, 14 Jul 2026 12:44:38 +0000 Subject: [PATCH 3/3] Address replication compression review --- .github/workflows/ci.yml | 2 +- tests/reduct/replication_api_test.cc | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fdfdf36..d726db9 100644 --- a/.github/workflows/ci.yml +++ b/.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]~[1_21]" + exclude_api_version_tag: "~[1_21]" audit_enabled: "false" steps: diff --git a/tests/reduct/replication_api_test.cc b/tests/reduct/replication_api_test.cc index 0f4c720..bf12305 100644 --- a/tests/reduct/replication_api_test.cc +++ b/tests/reduct/replication_api_test.cc @@ -215,7 +215,7 @@ TEST_CASE("reduct::Client should parse replication compression", "[replication_a } } -TEST_CASE("reduct::Client should set replication destination prefix", "[replication_api][1_20]") { +TEST_CASE("reduct::Client should set replication destination prefix", "[replication_api][1_21]") { Fixture ctx; auto settings = DefaultSettings(); settings.dst_prefix = "robot-1";