feat: support to force send non-idempotent write when doing duplication

src/common/duplication_common.cpp

@@ -37,6 +37,15 @@ DSN_DEFINE_uint32(replication,
                   "send mutation log batch bytes size per rpc");
 DSN_TAG_VARIABLE(duplicate_log_batch_bytes, FT_MUTABLE);
 
+DSN_DEFINE_bool(
+    replication,
+    duplication_unsafe_allow_non_idempotent,
+    false,
+    "Turn on the switch so that the cluster can accept non-idempotent writes and forward these "
+    "writes via duplication "
+    "Note that this switch may cause data inconsistency between clusters. So we say it is unsafe ");
+DSN_TAG_VARIABLE(duplication_unsafe_allow_non_idempotent, FT_MUTABLE);
+
 namespace dsn {
 namespace replication {
 

src/common/duplication_common.h

@@ -31,6 +31,7 @@
 #include "utils/fmt_utils.h"
 
 DSN_DECLARE_uint32(duplicate_log_batch_bytes);
+DSN_DECLARE_bool(duplication_unsafe_allow_non_idempotent);
 
 namespace dsn {
 namespace replication {

src/replica/duplication/mutation_batch.cpp

@@ -22,6 +22,7 @@
 #include <vector>
 
 #include "absl/strings/string_view.h"
+#include "common/duplication_common.h"
 #include "common/replication.codes.h"
 #include "consensus_types.h"
 #include "metadata_types.h"
@@ -173,7 +174,8 @@ void mutation_batch::add_mutation_if_valid(mutation_ptr &mu, decree start_decree
         // ERR_OPERATION_DISABLED, but there could still be a mutation written
         // before the duplication was added.
         // To ignore means this write will be lost, which is acceptable under this rare case.
-        if (!task_spec::get(update.code)->rpc_request_is_write_idempotent) {
+        if (!task_spec::get(update.code)->rpc_request_is_write_idempotent &&
+            !FLAGS_duplication_unsafe_allow_non_idempotent) {
             continue;
         }
         blob bb;

src/replica/mutation.h

@@ -48,6 +48,7 @@ class binary_reader;
 class binary_writer;
 class blob;
 class gpid;
+
 namespace utils {
 class latency_tracer;
 } // namespace utils

src/replica/replica_2pc.cpp

@@ -38,6 +38,7 @@
 
 #include "bulk_load/replica_bulk_loader.h"
 #include "bulk_load_types.h"
+#include "common/duplication_common.h"
 #include "common/fs_manager.h"
 #include "common/gpid.h"
 #include "common/replication.codes.h"
@@ -82,7 +83,6 @@ DSN_DEFINE_bool(replication,
                 true,
                 "reject client write requests if disk status is space insufficient");
 DSN_TAG_VARIABLE(reject_write_when_disk_insufficient, FT_MUTABLE);
-
 DSN_DEFINE_int32(replication,
                  prepare_timeout_ms_for_secondaries,
                  3000,
@@ -155,7 +155,8 @@ void replica::on_client_write(dsn::message_ex *request, bool ignore_throttling)
         return;
     }
 
-    if (is_duplication_master() && !spec->rpc_request_is_write_idempotent) {
+    if (is_duplication_master() && !spec->rpc_request_is_write_idempotent &&
+        !FLAGS_duplication_unsafe_allow_non_idempotent) {
         // Ignore non-idempotent write, because duplication provides no guarantee of atomicity to
         // make this write produce the same result on multiple clusters.
         METRIC_VAR_INCREMENT(dup_rejected_non_idempotent_write_requests);

src/server/info_collector.cpp

@@ -207,9 +207,11 @@ info_collector::app_stat_counters *info_collector::get_app_counters(const std::s
     INIT_COUNTER(incr_qps);
     INIT_COUNTER(check_and_set_qps);
     INIT_COUNTER(check_and_mutate_qps);
+    INIT_COUNTER(dup_unsafe_received_non_idempotent_duplicate_request);
     INIT_COUNTER(scan_qps);
     INIT_COUNTER(duplicate_qps);
     INIT_COUNTER(dup_shipped_ops);
+    INIT_COUNTER(dup_retry_non_idempotent_duplicate_request);
     INIT_COUNTER(dup_failed_shipping_ops);
     INIT_COUNTER(dup_recent_mutation_loss_count);
     INIT_COUNTER(recent_read_cu);

src/server/info_collector.h

@@ -66,10 +66,14 @@ class info_collector
             incr_qps->set(row_stats.incr_qps);
             check_and_set_qps->set(row_stats.check_and_set_qps);
             check_and_mutate_qps->set(row_stats.check_and_mutate_qps);
+            dup_unsafe_received_non_idempotent_duplicate_request->set(
+                row_stats.dup_unsafe_received_non_idempotent_duplicate_request);
             scan_qps->set(row_stats.scan_qps);
             duplicate_qps->set(row_stats.duplicate_qps);
             dup_shipped_ops->set(row_stats.dup_shipped_ops);
             dup_failed_shipping_ops->set(row_stats.dup_failed_shipping_ops);
+            dup_retry_non_idempotent_duplicate_request->set(
+                row_stats.dup_retry_non_idempotent_duplicate_request);
             dup_recent_mutation_loss_count->set(row_stats.dup_recent_mutation_loss_count);
             recent_read_cu->set(row_stats.recent_read_cu);
             recent_write_cu->set(row_stats.recent_write_cu);
@@ -144,10 +148,12 @@ class info_collector
         ::dsn::perf_counter_wrapper incr_qps;
         ::dsn::perf_counter_wrapper check_and_set_qps;
         ::dsn::perf_counter_wrapper check_and_mutate_qps;
+        ::dsn::perf_counter_wrapper dup_unsafe_received_non_idempotent_duplicate_request;
         ::dsn::perf_counter_wrapper scan_qps;
         ::dsn::perf_counter_wrapper duplicate_qps;
         ::dsn::perf_counter_wrapper dup_shipped_ops;
         ::dsn::perf_counter_wrapper dup_failed_shipping_ops;
+        ::dsn::perf_counter_wrapper dup_retry_non_idempotent_duplicate_request;
         ::dsn::perf_counter_wrapper dup_recent_mutation_loss_count;
         ::dsn::perf_counter_wrapper recent_read_cu;
         ::dsn::perf_counter_wrapper recent_write_cu;

src/server/pegasus_mutation_duplicator.cpp

@@ -48,6 +48,7 @@
 #include "utils/errors.h"
 #include "utils/fmt_logging.h"
 #include "utils/rand.h"
+#include "pegasus_rpc_types.h"
 
 METRIC_DEFINE_counter(replica,
                       dup_shipped_successful_requests,
@@ -59,6 +60,11 @@ METRIC_DEFINE_counter(replica,
                       dsn::metric_unit::kRequests,
                       "The number of failed DUPLICATE requests sent from client");
 
+METRIC_DEFINE_counter(replica,
+                      dup_retry_non_idempotent_duplicate_request,
+                      dsn::metric_unit::kRequests,
+                      "The number of retried non-idempotent DUPLICATE requests sent from client");
+
 namespace dsn {
 namespace replication {
 struct replica_base;
@@ -101,6 +107,22 @@ using namespace dsn::literals::chrono_literals;
         dsn::from_blob_to_thrift(data, thrift_request);
         return pegasus_hash_key_hash(thrift_request.hash_key);
     }
+    if (tc == dsn::apps::RPC_RRDB_RRDB_INCR) {
+        dsn::apps::incr_request thrift_request;
+        dsn::from_blob_to_thrift(data, thrift_request);
+        return pegasus_hash_key_hash(thrift_request.key);
+    }
+    if (tc == dsn::apps::RPC_RRDB_RRDB_CHECK_AND_SET) {
+        dsn::apps::check_and_set_request thrift_request;
+        dsn::from_blob_to_thrift(data, thrift_request);
+        return pegasus_hash_key_hash(thrift_request.hash_key);
+    }
+    if (tc == dsn::apps::RPC_RRDB_RRDB_CHECK_AND_MUTATE) {
+        dsn::apps::check_and_mutate_request thrift_request;
+        dsn::from_blob_to_thrift(data, thrift_request);
+        return pegasus_hash_key_hash(thrift_request.hash_key);
+    }
+
     LOG_FATAL("unexpected task code: {}", tc);
     __builtin_unreachable();
 }
@@ -189,6 +211,9 @@ void pegasus_mutation_duplicator::on_duplicate_reply(uint64_t hash,
             // retry this rpc
             _inflights[hash].push_front(rpc);
             _env.schedule([hash, cb, this]() { send(hash, cb); }, 1_s);
+
+            type_force_send_non_idempotent_if_need(rpc);
+
             return;
         }
         if (_inflights[hash].empty()) {
@@ -205,6 +230,58 @@ void pegasus_mutation_duplicator::on_duplicate_reply(uint64_t hash,
     }
 }
 
+void pegasus_mutation_duplicator::type_force_send_non_idempotent_if_need(duplicate_rpc &rpc)
+{
+    if (!FLAGS_duplication_unsafe_allow_non_idempotent) {
+        return;
+    }
+
+    // there maybe more than one mutation in one dup rpc
+    for (auto entry : rpc.request().entries) {
+        // not a non idempotent request
+        if (!_non_idempotent_code.count(entry.task_code)) {
+            continue;
+        }
+
+        METRIC_VAR_INCREMENT(dup_retry_non_idempotent_duplicate_request);
+        dsn::message_ex *write = dsn::from_blob_to_received_msg(entry.task_code, entry.raw_message);
+
+        if (entry.task_code == dsn::apps::RPC_RRDB_RRDB_INCR) {
+            incr_rpc raw_rpc(write);
+
+            LOG_DEBUG("Non-indempotent write RPC_RRDB_RRDB_INCR has been retried when doing "
+                      "duplication,"
+                      "key is [{}]",
+                      raw_rpc.request().key);
+            continue;
+        }
+
+        if (entry.task_code == dsn::apps::RPC_RRDB_RRDB_CHECK_AND_SET) {
+            check_and_set_rpc raw_rpc(write);
+
+            LOG_DEBUG("Non-indempotent write RPC_RRDB_RRDB_CHECK_AND_SET has been retried "
+                      "when doing duplication,"
+                      "hash key [{}], check sort key [{}],"
+                      "set sort key [{}]",
+                      raw_rpc.request().hash_key,
+                      raw_rpc.request().check_sort_key,
+                      raw_rpc.request().set_sort_key);
+            continue;
+        }
+
+        if (entry.task_code == dsn::apps::RPC_RRDB_RRDB_CHECK_AND_MUTATE) {
+            check_and_mutate_rpc raw_rpc(write);
+
+            LOG_DEBUG("Non-indempotent write RPC_RRDB_RRDB_CHECK_AND_MUTATE has been "
+                      "retried when doing duplication,"
+                      "hash key is [{}] , sort key is [{}] .",
+                      raw_rpc.request().hash_key,
+                      raw_rpc.request().check_sort_key);
+            continue;
+        }
+    }
+}
+
 void pegasus_mutation_duplicator::duplicate(mutation_tuple_set muts, callback cb)
 {
     _total_shipped_size = 0;

src/server/pegasus_mutation_duplicator.h

@@ -23,21 +23,24 @@
 #include <stdint.h>
 #include <deque>
 #include <map>
+#include <set>
 #include <string>
 
+#include "absl/strings/string_view.h"
 #include "replica/duplication/mutation_duplicator.h"
 #include "rrdb/rrdb.client.h"
+#include "rrdb/rrdb.code.definition.h"
 #include "runtime/pipeline.h"
 #include "runtime/task/task_code.h"
 #include "runtime/task/task_tracker.h"
 #include "utils/chrono_literals.h"
-#include "absl/strings/string_view.h"
 #include "utils/metrics.h"
 #include "utils/zlocks.h"
 
 namespace dsn {
 class blob;
 class error_code;
+
 namespace replication {
 struct replica_base;
 } // namespace replication
@@ -73,6 +76,8 @@ class pegasus_mutation_duplicator : public dsn::replication::mutation_duplicator
 
     void on_duplicate_reply(uint64_t hash, callback, duplicate_rpc, dsn::error_code err);
 
+    void type_force_send_non_idempotent_if_need(duplicate_rpc &rpc);
+
 private:
     friend class pegasus_mutation_duplicator_test;
 
@@ -89,8 +94,14 @@ class pegasus_mutation_duplicator : public dsn::replication::mutation_duplicator
 
     size_t _total_shipped_size{0};
 
+    const std::set<dsn::task_code> _non_idempotent_code = {
+        dsn::apps::RPC_RRDB_RRDB_INCR,
+        dsn::apps::RPC_RRDB_RRDB_CHECK_AND_SET,
+        dsn::apps::RPC_RRDB_RRDB_CHECK_AND_MUTATE};
+
     METRIC_VAR_DECLARE_counter(dup_shipped_successful_requests);
     METRIC_VAR_DECLARE_counter(dup_shipped_failed_requests);
+    METRIC_VAR_DECLARE_counter(dup_retry_non_idempotent_duplicate_request);
 };
 
 // Decodes the binary `request_data` into write request in thrift struct, and

src/server/pegasus_write_service.cpp

@@ -121,6 +121,11 @@ METRIC_DEFINE_counter(replica,
                       dsn::metric_unit::kRequests,
                       "The number of DUPLICATE requests");
 
+METRIC_DEFINE_counter(replica,
+                      dup_unsafe_received_non_idempotent_duplicate_request,
+                      dsn::metric_unit::kRequests,
+                      "statistic the those no idempotent qps of DUPLICATE requests Force received");
+
 METRIC_DEFINE_percentile_int64(replica,
                                dup_time_lag_ms,
                                dsn::metric_unit::kMilliSeconds,
@@ -168,6 +173,7 @@ pegasus_write_service::pegasus_write_service(pegasus_server_impl *server)
       METRIC_VAR_INIT_replica(check_and_set_latency_ns),
       METRIC_VAR_INIT_replica(check_and_mutate_latency_ns),
       METRIC_VAR_INIT_replica(dup_requests),
+      METRIC_VAR_INIT_replica(dup_unsafe_received_non_idempotent_duplicate_request),
       METRIC_VAR_INIT_replica(dup_time_lag_ms),
       METRIC_VAR_INIT_replica(dup_lagging_writes),
       _put_batch_size(0),
@@ -415,6 +421,44 @@ int pegasus_write_service::duplicate(int64_t decree,
             }
             continue;
         }
+
+        // Parse non-idempotent writes via duplication
+        if (request.task_code == dsn::apps::RPC_RRDB_RRDB_INCR ||
+            request.task_code == dsn::apps::RPC_RRDB_RRDB_CHECK_AND_SET ||
+            request.task_code == dsn::apps::RPC_RRDB_RRDB_CHECK_AND_MUTATE) {
+            // receive non-idempotent request from master cluster via duplication when
+            // FLAG_duplication_unsafe_allow_non_idempotent set as true.
+            // This metric greater than zero means that there is already the possibility of
+            // inconsistency between clusters
+            METRIC_VAR_INCREMENT(dup_unsafe_received_non_idempotent_duplicate_request);
+
+            if (request.task_code == dsn::apps::RPC_RRDB_RRDB_INCR) {
+                incr_rpc rpc(write);
+                resp.__set_error(_impl->incr(ctx.decree, rpc.request(), rpc.response()));
+                if (resp.error != rocksdb::Status::kOk) {
+                    return resp.error;
+                }
+                continue;
+            }
+            if (request.task_code == dsn::apps::RPC_RRDB_RRDB_CHECK_AND_SET) {
+                check_and_set_rpc rpc(write);
+                resp.__set_error(_impl->check_and_set(ctx.decree, rpc.request(), rpc.response()));
+                if (resp.error != rocksdb::Status::kOk) {
+                    return resp.error;
+                }
+                continue;
+            }
+            if (request.task_code == dsn::apps::RPC_RRDB_RRDB_CHECK_AND_MUTATE) {
+                check_and_mutate_rpc rpc(write);
+                resp.__set_error(
+                    _impl->check_and_mutate(ctx.decree, rpc.request(), rpc.response()));
+                if (resp.error != rocksdb::Status::kOk) {
+                    return resp.error;
+                }
+                continue;
+            }
+        }
+
         resp.__set_error(rocksdb::Status::kInvalidArgument);
         resp.__set_error_hint(fmt::format("unrecognized task code {}", request.task_code));
         return empty_put(ctx.decree);

src/server/pegasus_write_service.h

@@ -232,6 +232,7 @@ class pegasus_write_service : dsn::replication::replica_base
     METRIC_VAR_DECLARE_percentile_int64(check_and_mutate_latency_ns);
 
     METRIC_VAR_DECLARE_counter(dup_requests);
+    METRIC_VAR_DECLARE_counter(dup_unsafe_received_non_idempotent_duplicate_request);
     METRIC_VAR_DECLARE_percentile_int64(dup_time_lag_ms);
     METRIC_VAR_DECLARE_counter(dup_lagging_writes);
 

src/server/test/pegasus_mutation_duplicator_test.cpp

@@ -23,7 +23,6 @@
 #include <fmt/core.h>
 #include <pegasus/error.h>
 #include <sys/types.h>
-#include <algorithm>
 #include <memory>
 #include <tuple>
 #include <utility>