feat(split): add partition_version

include/dsn/dist/replication/replication_app_base.h

@@ -228,6 +228,8 @@ class replication_app_base : public replica_base
     // query app envs.
     virtual void query_app_envs(/*out*/ std::map<std::string, std::string> &envs) = 0;
 
+    virtual void set_partition_version(int32_t partition_version) = 0;
+
 public:
     //
     // utility functions to be used by app

src/dist/replication/lib/replica.cpp

@@ -65,6 +65,7 @@ replica::replica(
     _options = &stub->options();
     init_state();
     _config.pid = gpid;
+    _partition_version = app.partition_count - 1;
 
     std::string counter_str = fmt::format("private.log.size(MB)@{}", gpid);
     _counter_private_log_size.init_app_counter(

src/dist/replication/lib/replica.h

@@ -494,6 +494,9 @@ class replica : public serverlet<replica>, public ref_counter, public replica_ba
     // ballot when starting partition split and split will stop if ballot changed
     // _child_init_ballot = 0 if partition not in partition split
     ballot _child_init_ballot{0};
+    // in normal case _partition_version = partition_count-1
+    // when parent register child, _partition_version = -1, will reject client read write request
+    std::atomic<int32_t> _partition_version;
 
     // perf counters
     perf_counter_wrapper _counter_private_log_size;

src/dist/replication/lib/replica_init.cpp

@@ -312,6 +312,7 @@ error_code replica::init_app_and_prepare_list(bool create_new)
         _app = nullptr;
     } else {
         _is_initializing = true;
+        _app->set_partition_version(_app_info.partition_count - 1);
 
         if (nullptr == _private_log) {
             ddebug("%s: clear private log, dir = %s", name(), log_dir.c_str());

src/dist/replication/storage_engine/simple_kv/simple_kv.server.impl.h

@@ -90,6 +90,8 @@ class simple_kv_service_impl : public simple_kv_service
 
     virtual void query_app_envs(/*out*/ std::map<std::string, std::string> &envs) {}
 
+    virtual void set_partition_version(int32_t partition_version) {}
+
 private:
     void recover();
     void recover(const std::string &name, int64_t version);

src/dist/replication/test/replica_test/unit_test/mock_utils.h

@@ -69,6 +69,8 @@ class mock_replication_app_base : public replication_app_base
     void query_app_envs(std::map<std::string, std::string> &out) override { out = _envs; }
     decree last_durable_decree() const override { return 0; }
 
+    void set_partition_version(int32_t partition_version) override {}
+
 private:
     std::map<std::string, std::string> _envs;
     decree _decree = 5;

src/dist/replication/test/simple_kv/simple_kv.server.impl.h

@@ -90,6 +90,8 @@ class simple_kv_service_impl : public application::simple_kv_service
 
     virtual void query_app_envs(/*out*/ std::map<std::string, std::string> &envs) {}
 
+    virtual void set_partition_version(int32_t partition_version) {}
+
 private:
     void recover();
     void recover(const std::string &name, int64_t version);