feat(dup): implement server handling of duplicate rpc (part 1) (#456)

apache · Jan 13, 2020 · c5dec62 · c5dec62
1 parent 0268530
commit c5dec62
Show file tree

Hide file tree

Showing 11 changed files with 265 additions and 41 deletions.
diff --git a/src/base/pegasus_value_schema.h b/src/base/pegasus_value_schema.h
@@ -21,6 +21,12 @@ namespace pegasus {
 
 #define PEGASUS_DATA_VERSION_MAX 0u
 
+/// Generates timetag in host endian.
+inline uint64_t generate_timetag(uint64_t timestamp, uint8_t cluster_id, bool delete_tag)
+{
+    return timestamp << 8u | cluster_id << 1u | delete_tag;
+}
+
 /// Extracts expire_ts from rocksdb value with given version.
 /// The value schema must be in v0.
 /// \return expire_ts in host endian
@@ -46,7 +52,7 @@ pegasus_extract_user_data(uint32_t version, std::string &&raw_value, ::dsn::blob
               version,
               PEGASUS_DATA_VERSION_MAX);
 
-    std::string *s = new std::string(std::move(raw_value));
+    auto *s = new std::string(std::move(raw_value));
     dsn::data_input input(*s);
     input.skip(sizeof(uint32_t));
     dsn::string_view view = input.read_str();
@@ -127,7 +133,7 @@ class pegasus_value_generator
             _write_slices.emplace_back(user_data.data(), user_data.length());
         }
 
-        return rocksdb::SliceParts(&_write_slices[0], static_cast<int>(_write_slices.size()));
+        return {&_write_slices[0], static_cast<int>(_write_slices.size())};
     }
 
 private:

diff --git a/src/base/rrdb_types.cpp b/src/base/rrdb_types.cpp
diff --git a/src/idl/rrdb.thrift b/src/idl/rrdb.thrift
@@ -264,11 +264,17 @@ struct duplicate_request
 
     // ID of the cluster where this write comes from.
     4: optional byte cluster_id
+
+    // Whether to compare the timetag of old value with the new write's.
+    5: optional bool verify_timetag
 }
 
 struct duplicate_response
 {
     1: optional i32 error;
+
+    // hints on the reason why this duplicate failed.
+    2: optional string error_hint;
 }
 
 service rrdb

diff --git a/src/include/rrdb/rrdb_types.h b/src/include/rrdb/rrdb_types.h
diff --git a/src/server/pegasus_mutation_duplicator.cpp b/src/server/pegasus_mutation_duplicator.cpp
@@ -135,6 +135,8 @@ void pegasus_mutation_duplicator::on_duplicate_reply(uint64_t hash,
                            err == dsn::ERR_OK ? _client->get_error_string(perr) : err.to_string(),
                            rpc.request().timestamp);
         }
+        // duplicating an illegal write to server is unacceptable, fail fast.
+        dassert_replica(perr != PERR_INVALID_ARGUMENT, rpc.response().error_hint);
     } else {
         _shipped_ops->increment();
         _total_shipped_size +=