SDSTOR-22816: reject duplicate create_pg while inflight

conanfile.py

@@ -10,7 +10,7 @@
 
 class HomeObjectConan(ConanFile):
     name = "homeobject"
-    version = "4.3.5"
+    version = "4.3.6"
 
     homepage = "https://github.com/eBay/HomeObject"
     description = "Blob Store built on HomeStore"

src/lib/homestore_backend/hs_homeobject.hpp

@@ -2,6 +2,7 @@
 
 #include <memory>
 #include <mutex>
+#include <unordered_set>
 
 #include <homestore/homestore.hpp>
 #include <homestore/index/index_table.hpp>
@@ -111,6 +112,9 @@ class HSHomeObject : public HomeObjectImpl {
     // Shard migration info: tracks shards that need migration from v1 to v2 format
     std::vector< shard_id_t > shards_to_migrate_;
     std::vector< pg_id_t > destroyed_stale_pgs_;
+    // pg ids whose create_pg has passed admission but not finished. A CM retry that arrives
+    // in this window must not allocate a second repldev for the same pg.
+    std::unordered_set< pg_id_t > creating_pgs_;
 
 public:
     // Old version shard_info_superblk (v0.02) - for backward compatibility testing and migration

src/lib/homestore_backend/hs_pg_manager.cpp

@@ -91,29 +91,55 @@ PGManager::NullAsyncResult HSHomeObject::_create_pg(PGInfo&& pg_info, std::set<
     incr_pending_request_num();
 
     auto pg_id = pg_info.id;
-    auto hs_pg = get_hs_pg(pg_id);
-    if (hs_pg) {
-        if (!pg_info.is_equivalent_to(hs_pg->pg_info_)) {
-            LOGW("PG already exists with different info! pg={}, pg_info={}, hs_pg_info={}", pg_id, pg_info.to_string(),
-                 hs_pg->pg_info_.to_string());
+    // Admit under _pg_lock together with the _pg_map lookup. Otherwise two create_pg calls can both
+    // observe "not in map" and each create a repldev. The later failure then returns this pg's chunks.
+    const HS_PG* existing_pg = nullptr;
+    {
+        std::unique_lock lck(_pg_lock);
+        existing_pg = _get_hs_pg_unlocked(pg_id);
+        if (existing_pg == nullptr) {
+            if (creating_pgs_.contains(pg_id)) {
+                LOGW("PG create is in progress, retry later, pg={}, trace_id={}", pg_id, tid);
+                decr_pending_request_num();
+                return folly::makeUnexpected(PGError::RETRY_REQUEST);
+            }
+            creating_pgs_.insert(pg_id);
+        }
+    }
+    if (existing_pg) {
+        if (!pg_info.is_equivalent_to(existing_pg->pg_info_)) {
+            LOGW("PG already exists with different info! pg={}, trace_id={}, pg_info={}, hs_pg_info={}", pg_id, tid,
+                 pg_info.to_string(), existing_pg->pg_info_.to_string());
             decr_pending_request_num();
             return folly::makeUnexpected(PGError::INVALID_ARG);
         }
-        LOGW("PG already exists! pg={}", pg_id);
+        LOGW("PG already exists! pg={}, trace_id={}", pg_id, tid);
         decr_pending_request_num();
         return folly::Unit();
     }
 
+    auto finish_creating_pg = [this, pg_id]() {
+        std::unique_lock lck(_pg_lock);
+        creating_pgs_.erase(pg_id);
+    };
+
+#ifdef _PRERELEASE
+    // Test gate, after the pg is marked in-flight. No-op when the flip is not injected.
+    iomgr_flip::instance()->callback_flip("simulate_create_pg_delay");
+#endif
+
     const auto chunk_size = chunk_selector()->get_chunk_size();
     if (pg_info.size < chunk_size) {
-        LOGW("Not support to create PG which pg_size={} < chunk_size={}", pg_info.size, chunk_size);
+        LOGW("Not support to create PG which pg_size={} < chunk_size={}, trace_id={}", pg_info.size, chunk_size, tid);
+        finish_creating_pg();
         decr_pending_request_num();
         return folly::makeUnexpected(PGError::INVALID_ARG);
     }
 
     auto const num_chunk = chunk_selector()->select_chunks_for_pg(pg_id, pg_info.size);
     if (!num_chunk.has_value()) {
-        LOGW("Failed to select chunks for pg={}", pg_id);
+        LOGW("Failed to select chunks for pg={}, trace_id={}", pg_id, tid);
+        finish_creating_pg();
         decr_pending_request_num();
         return folly::makeUnexpected(PGError::NO_SPACE_LEFT);
     }
@@ -139,7 +165,7 @@ PGManager::NullAsyncResult HSHomeObject::_create_pg(PGInfo&& pg_info, std::set<
             // FIXME:https://github.com/eBay/HomeObject/pull/136#discussion_r1470504271
             return do_create_pg(v.value(), std::move(pg_info), tid);
         })
-        .thenValue([this, pg_id, repl_dev_group_id, tid](auto&& r) -> PGManager::NullAsyncResult {
+        .thenValue([this, pg_id, repl_dev_group_id, tid, finish_creating_pg](auto&& r) -> PGManager::NullAsyncResult {
             // reclaim resources if failed to create pg
             if (r.hasError()) {
                 bool res = chunk_selector_->return_pg_chunks_to_dev_heap(pg_id);
@@ -148,15 +174,21 @@ PGManager::NullAsyncResult HSHomeObject::_create_pg(PGInfo&& pg_info, std::set<
                 // if don't have repl dev, it will return ReplServiceError::SERVER_NOT_FOUND
                 return hs_repl_service()
                     .remove_repl_dev(repl_dev_group_id)
-                    .deferValue([r, repl_dev_group_id, this](auto&& e) -> PGManager::NullAsyncResult {
+                    .deferValue([r, repl_dev_group_id, tid, this,
+                                 finish_creating_pg](auto&& e) -> PGManager::NullAsyncResult {
                         if (e != ReplServiceError::OK) {
-                            LOGW("Failed to remove repl device which group_id={}, error={}", repl_dev_group_id, e);
+                            LOGW("Failed to remove repl device which group_id={}, error={}, trace_id={}",
+                                 repl_dev_group_id, e, tid);
                         }
+                        // Drop the inflight mark only after chunk/repldev cleanup, so a retry cannot
+                        // select the same chunks while they are being returned.
+                        finish_creating_pg();
                         decr_pending_request_num();
                         // still return the original error
                         return folly::makeUnexpected(r.error());
                     });
             }
+            finish_creating_pg();
             decr_pending_request_num();
             return folly::Unit();
         });

src/lib/homestore_backend/tests/hs_pg_tests.cpp

@@ -1,6 +1,8 @@
 #include "homeobj_fixture.hpp"
 #include <homestore/replication_service.hpp>
 
+#include <future>
+
 TEST_F(HomeObjectFixture, PGStatsTest) {
     LOGINFO("HomeObject replica={} setup completed", g_helper->replica_num());
     //  Create a pg, shard, put blob should succeed.
@@ -359,6 +361,49 @@ TEST_F(HomeObjectFixture, DuplicateCreatePG) {
 }
 
 #ifdef _PRERELEASE
+// First create is held inside _create_pg after creating_pgs_ insert.
+// A duplicate during that window returns RETRY_REQUEST. After first.get() the pg is in _pg_map, so the next
+// duplicate succeeds.
+TEST_F(HomeObjectFixture, CreatePGRetryWhileInflight) {
+    if (0 != g_helper->replica_num()) { return; }
+
+    pg_id_t pg_id{1};
+    std::promise< void > entered;
+    std::promise< void > release;
+    set_callback_flip("simulate_create_pg_delay", [&]() {
+        entered.set_value();
+        release.get_future().wait();
+    });
+
+    auto members = g_helper->members();
+    auto name = g_helper->test_name();
+    auto pg_size = SISL_OPTIONS["chunks_per_pg"].as< uint64_t >() * SISL_OPTIONS["chunk_size"].as< uint64_t >() * Mi;
+    auto make_info = [&]() {
+        auto info = homeobject::PGInfo(pg_id);
+        info.size = pg_size;
+        for (const auto& member : members) {
+            info.members.insert(homeobject::PGMember{member.first, name + std::to_string(member.second),
+                                                     member.second == 0 ? 1 : 0});
+        }
+        return info;
+    };
+
+    auto first =
+        std::async(std::launch::async, [&]() { return _obj_inst->pg_manager()->create_pg(make_info()).get(); });
+    entered.get_future().wait();
+
+    auto retry = _obj_inst->pg_manager()->create_pg(make_info()).get();
+    // Release before any fatal assertion; otherwise ~future of `first` blocks forever.
+    release.set_value();
+    ASSERT_FALSE(retry);
+    ASSERT_EQ(PGError::RETRY_REQUEST, retry.error());
+
+    ASSERT_TRUE(first.get());
+
+    auto again = _obj_inst->pg_manager()->create_pg(make_info()).get();
+    ASSERT_TRUE(again);
+}
+
 TEST_F(HomeObjectFixture, CreatePGFailed) {
     set_basic_flip("create_pg_create_repl_dev_error", 1); // simulate create pg repl dev error
     set_basic_flip("create_pg_raft_message_error", 1);    // simulate create pg raft message error