Fix pg membership inconsistency (#386)

- Implement on_pg_clean_replace_member_task callback which actually rollback the membership change
- Add a reconcile_membership API which is used to correct the existing inconsistency problem.

Co-authored-by: yuwmao <yuwmao@ebaychina.com>

Author: yuwmao

conanfile.py

@@ -10,7 +10,7 @@
 
 class HomeObjectConan(ConanFile):
     name = "homeobject"
-    version = "3.0.17"
+    version = "3.0.18"
 
     homepage = "https://github.com/eBay/HomeObject"
     description = "Blob Store built on HomeStore"
@@ -50,7 +50,7 @@ def build_requirements(self):
 
     def requirements(self):
         self.requires("sisl/[^13.0]@oss/master", transitive_headers=True)
-        self.requires("homestore/[^7.1]@oss/master")
+        self.requires("homestore/[^7.3]@oss/master")
         self.requires("iomgr/[^12.0]@oss/master")
 
     def validate(self):

src/lib/homestore_backend/hs_homeobject.hpp

@@ -789,8 +789,26 @@ class HSHomeObject : public HomeObjectImpl {
                                        const homestore::replica_member_info& member_out,
                                        const homestore::replica_member_info& member_in, trace_id_t tid);
 
+    /**
+     * @brief Called when clean replace member task (rollback)
+     * @param group_id Group ID
+     * @param task_id Task ID
+     * @param member_out Member which should be restored to group
+     * @param member_in Member which should be removed from group
+     * */
+    void on_pg_clean_replace_member_task(homestore::group_id_t group_id, const std::string& task_id,
+                                         const homestore::replica_member_info& member_out,
+                                         const homestore::replica_member_info& member_in, trace_id_t tid);
+
     void on_remove_member(homestore::group_id_t group_id, const peer_id_t& member, trace_id_t tid = 0);
 
+    /**
+     * @brief Reconcile PG membership by synchronizing with raft's actual member list
+     * @param pg_id PG ID to reconcile
+     * @return true if reconciled successfully, false otherwise
+     */
+    bool reconcile_membership(pg_id_t pg_id);
+
     /**
      * @brief Cleans up and recycles resources for the PG identified by the given pg_id on the current node.
      *

src/lib/homestore_backend/hs_http_manager.cpp

@@ -48,6 +48,8 @@ HttpManager::HttpManager(HSHomeObject& ho) : ho_(ho) {
          Pistache::Rest::Routes::bind(&HttpManager::clean_replace_member_task, this)},
         {Pistache::Http::Method::Get, "/api/v1/pg_replacemember_tasks",
          Pistache::Rest::Routes::bind(&HttpManager::list_pg_replace_member_task, this)},
+        {Pistache::Http::Method::Post, "/api/v1/reconcile_membership",
+         Pistache::Rest::Routes::bind(&HttpManager::reconcile_membership, this)},
         {Pistache::Http::Method::Delete, "/api/v1/pg", Pistache::Rest::Routes::bind(&HttpManager::exit_pg, this)},
 #ifdef _PRERELEASE
         {Pistache::Http::Method::Post, "/api/v1/crashSystem",
@@ -333,6 +335,35 @@ void HttpManager::clean_replace_member_task(const Pistache::Rest::Request& reque
         response.send(Pistache::Http::Code::Bad_Request, std::string("Invalid JSON: ") + e.what());
     }
 }
+void HttpManager::reconcile_membership(const Pistache::Rest::Request& request,
+                                       Pistache::Http::ResponseWriter response) {
+    try {
+        auto body = request.body();
+        auto j = nlohmann::json::parse(body);
+
+        std::string pg_id_str = j.at("pg_id").get< std::string >();
+        pg_id_t pg_id = std::stoull(pg_id_str);
+
+        LOGINFO("Reconcile membership for pg_id={}", pg_id);
+
+        bool success = ho_.reconcile_membership(pg_id);
+        if (!success) {
+            response.send(Pistache::Http::Code::Internal_Server_Error,
+                          fmt::format("Failed to reconcile membership for pg_id={}", pg_id));
+            return;
+        }
+
+        nlohmann::json result;
+        result["status"] = "success";
+        result["pg_id"] = pg_id;
+        result["message"] = "Membership reconciled successfully";
+
+        response.send(Pistache::Http::Code::Ok, result.dump());
+    } catch (const std::exception& e) {
+        response.send(Pistache::Http::Code::Bad_Request, std::string("Invalid JSON: ") + e.what());
+    }
+}
+
 void HttpManager::list_pg_replace_member_task(const Pistache::Rest::Request& request,
                                               Pistache::Http::ResponseWriter response) {
     auto tid = generateRandomTraceId();

src/lib/homestore_backend/hs_http_manager.hpp

@@ -37,6 +37,7 @@ class HttpManager {
     void remove_member(const Pistache::Rest::Request& request, Pistache::Http::ResponseWriter response);
     void clean_replace_member_task(const Pistache::Rest::Request& request, Pistache::Http::ResponseWriter response);
     void list_pg_replace_member_task(const Pistache::Rest::Request& request, Pistache::Http::ResponseWriter response);
+    void reconcile_membership(const Pistache::Rest::Request& request, Pistache::Http::ResponseWriter response);
     void get_pg_quorum(const Pistache::Rest::Request& request, Pistache::Http::ResponseWriter response);
     void exit_pg(const Pistache::Rest::Request& request, Pistache::Http::ResponseWriter response);
 

src/lib/homestore_backend/hs_pg_manager.cpp

@@ -398,6 +398,108 @@ void HSHomeObject::on_pg_complete_replace_member(group_id_t group_id, const std:
          boost::uuids::to_string(member_out.id), boost::uuids::to_string(member_in.id), tid);
 }
 
+//This function actually perform rollback for replace member task:  Remove in_member, and ensure out_member exists
+void HSHomeObject::on_pg_clean_replace_member_task(group_id_t group_id, const std::string& task_id,
+                                                    const replica_member_info& member_out,
+                                                    const replica_member_info& member_in, trace_id_t tid) {
+    std::unique_lock lck(_pg_lock);
+    for (const auto& iter : _pg_map) {
+        auto& pg = iter.second;
+        if (pg_repl_dev(*pg).group_id() == group_id) {
+            auto hs_pg = static_cast< HSHomeObject::HS_PG* >(pg.get());
+
+            // Remove the in_member (the one that was added but now needs to be removed)
+            auto removed_count = pg->pg_info_.members.erase(PGMember(member_in.id));
+
+            // Ensure out_member exists
+            // Using emplace is safe - it won't overwrite if already exists
+            auto out_pg_member = to_pg_member(member_out);
+            auto [it, inserted] = pg->pg_info_.members.emplace(std::move(out_pg_member));
+
+            // Update superblock
+            hs_pg->update_membership(pg->pg_info_.members);
+
+            LOGI("PG clean replace member task done (rollback), task_id={}, removed in_member={} (removed={}), "
+                 "ensured out_member={} (inserted={}), member_nums={}, trace_id={}",
+                 task_id, boost::uuids::to_string(member_in.id), removed_count,
+                 boost::uuids::to_string(member_out.id), inserted, pg->pg_info_.members.size(), tid);
+            return;
+        }
+    }
+    LOGE("PG clean replace member task failed, pg not found, task_id={}, member_out={}, member_in={}, trace_id={}", task_id,
+         boost::uuids::to_string(member_out.id), boost::uuids::to_string(member_in.id), tid);
+}
+
+bool HSHomeObject::reconcile_membership(pg_id_t pg_id) {
+    std::unique_lock lck(_pg_lock);
+    auto hs_pg = const_cast< HS_PG* >(_get_hs_pg_unlocked(pg_id));
+    if (!hs_pg) {
+        LOGE("PG {} not found for reconcile_membership", pg_id);
+        return false;
+    }
+
+    // Get actual members from replication layer
+    auto& repl_dev = hs_pg->repl_dev_;
+    auto actual_members = repl_dev->get_replication_quorum();
+    if (actual_members.empty()) {
+        LOGW("Failed to get replication quorum for PG {}, repl_dev may not be ready", pg_id);
+        return false;
+    }
+
+    // Build new member set from actual members
+    std::set< PGMember > new_members;
+    for (auto& member_id : actual_members) {
+        auto existing = hs_pg->pg_info_.members.find(PGMember(member_id));
+        if (existing != hs_pg->pg_info_.members.end()) {
+            // Keep the existing member with its name and priority
+            new_members.insert(*existing);
+        } else {
+            // New member not in our records, add with default name
+            PGMember new_member(member_id);
+            new_members.insert(std::move(new_member));
+            LOGE("Adding new member {} to pg={} membership, should not happen!", boost::uuids::to_string(member_id), hs_pg->pg_info_.id);
+        }
+    }
+    // Check if membership changed
+    if (new_members == hs_pg->pg_info_.members) {
+        LOGD("Membership already in sync for pg={}, no update needed", hs_pg->pg_info_.id);
+        return true;
+    }
+
+    // Log the differences
+    std::vector< peer_id_t > removed_members;
+    std::vector< peer_id_t > added_members;
+
+    for (auto& old_member : hs_pg->pg_info_.members) {
+        if (new_members.find(old_member) == new_members.end()) {
+            removed_members.push_back(old_member.id);
+        }
+    }
+
+    for (auto& new_member : new_members) {
+        if (hs_pg->pg_info_.members.find(new_member) == hs_pg->pg_info_.members.end()) {
+            added_members.push_back(new_member.id);
+        }
+    }
+
+    LOGI("Reconciling PG {} membership: removing {} members, adding {} members, old membership count {}, new membership count: {}",
+         pg_id, removed_members.size(), added_members.size(), hs_pg->pg_info_.members.size(), new_members.size());
+
+    for (auto& member_id : removed_members) {
+        LOGI("  - Removing member: {}", boost::uuids::to_string(member_id));
+    }
+    for (auto& member_id : added_members) {
+        LOGI("  - Adding member: {}", boost::uuids::to_string(member_id));
+    }
+
+    // Update membership
+    hs_pg->pg_info_.members = std::move(new_members);
+    hs_pg->update_membership(hs_pg->pg_info_.members);
+
+    LOGI("Successfully reconciled PG {} membership, new member_count={}", pg_id, hs_pg->pg_info_.members.size());
+    return true;
+}
+
 void HSHomeObject::on_remove_member(homestore::group_id_t group_id, const peer_id_t& member, trace_id_t tid) {
     LOGINFO("PG remove member, member={}, trace_id={}", boost::uuids::to_string(member), tid);
     std::unique_lock lck(_pg_lock);

src/lib/homestore_backend/replication_state_machine.cpp

@@ -274,6 +274,13 @@ void ReplicationStateMachine::on_complete_replace_member(const std::string& task
     home_object_->on_pg_complete_replace_member(repl_dev()->group_id(), task_id, member_out, member_in, tid);
 }
 
+void ReplicationStateMachine::on_clean_replace_member_task(const std::string& task_id,
+                                                           const homestore::replica_member_info& member_out,
+                                                           const homestore::replica_member_info& member_in,
+                                                           trace_id_t tid) {
+    home_object_->on_pg_clean_replace_member_task(repl_dev()->group_id(), task_id, member_out, member_in, tid);
+}
+
 void ReplicationStateMachine::on_destroy(const homestore::group_id_t& group_id) {
     auto PG_ID = home_object_->get_pg_id_with_group_id(group_id);
     if (!PG_ID.has_value()) {

src/lib/homestore_backend/replication_state_machine.hpp

@@ -183,6 +183,10 @@ class ReplicationStateMachine : public homestore::ReplDevListener {
     void on_complete_replace_member(const std::string& task_id, const homestore::replica_member_info& member_out,
                                     const homestore::replica_member_info& member_in, trace_id_t tid = 0) override;
 
+    /// @brief Called when clean replace member task (rollback)
+    void on_clean_replace_member_task(const std::string& task_id, const homestore::replica_member_info& member_out,
+                                      const homestore::replica_member_info& member_in, trace_id_t tid = 0) override;
+
     /// @brief Called when the replica is being destroyed by nuraft;
     void on_destroy(const homestore::group_id_t& group_id) override;