fix some review comments in create_shard

src/lib/homeobject_impl.hpp

@@ -20,6 +20,7 @@ struct Shard {
     explicit Shard(ShardInfo info) : info(std::move(info)) {}
     ShardInfo info;
     uint16_t chunk_id;
+    void* metablk_cookie{nullptr};
 };
 
 using ShardList = std::list< Shard >;
@@ -47,7 +48,7 @@ class HomeObjectImpl : public HomeObject,
     virtual BlobManager::NullResult _del_blob(ShardInfo const&, blob_id) = 0;
     ///
     folly::Future< ShardManager::Result< Shard > > _get_shard(shard_id id) const;
-    auto _defer() const { return folly::makeSemiFuture().via(folly::getGlobalCPUExecutor()); }
+    auto _defer() const { return folly::makeSemiFuture().via(folly::getGlobalCPUExecutor());}
 
 protected:
     std::mutex _repl_lock;
@@ -63,7 +64,7 @@ class HomeObjectImpl : public HomeObject,
     std::map< pg_id, PG > _pg_map;
 
     mutable std::shared_mutex _shard_lock;
-    std::map < shard_id, ShardIterator > _shard_map;
+    std::map< shard_id, ShardIterator > _shard_map;
     ///
     PGManager::Result< PG > _get_pg(pg_id pg);
 public:

src/lib/homestore/homeobject.cpp

@@ -94,12 +94,11 @@ void HSHomeObject::on_shard_meta_blk_found(homestore::meta_blk* mblk, sisl::byte
     shard_info_str.append(r_cast<const char*>(buf.bytes()), size);
 
     auto shard = deserialize_shard(shard_info_str);
-    if (shard.info.state == ShardInfo::State::OPEN) {
-        // create shard;
-        do_commit_new_shard(shard);
-    } else {
-        do_commit_seal_shard(shard);
-    }
+    shard.metablk_cookie = mblk;
+
+    //As shard info in the homestore metablk is always the latest state(OPEN or SEALED),
+    //we can always create a shard from this shard info and once shard is deleted, the associated metablk will be deleted too.
+    do_commit_new_shard(shard);
 }
 
 } // namespace homeobject

src/lib/homestore/homeobject.hpp

@@ -33,6 +33,7 @@ class HSHomeObject : public HomeObjectImpl {
     void do_commit_new_shard(const Shard& shard);
     void do_commit_seal_shard(const Shard& shard);
     void register_homestore_metablk_callback();
+    void* get_shard_metablk(shard_id id);
 public:
     using HomeObjectImpl::HomeObjectImpl;
     ~HSHomeObject();

src/lib/homestore/shard_manager.cpp

@@ -87,7 +87,7 @@ ShardManager::Result< ShardInfo > HSHomeObject::_create_shard(pg_id pg_owner, ui
     //preapre msg header;
     const uint32_t needed_size = sizeof(ReplicationMessageHeader) + create_shard_message.size();
     auto buf = nuraft::buffer::alloc(needed_size);
-    uint8_t* raw_ptr = static_cast<uint8_t*>(buf->data_begin());
+    uint8_t* raw_ptr = buf->data_begin();
     ReplicationMessageHeader *header = new(raw_ptr) ReplicationMessageHeader();
     header->message_type = ReplicationMessageType::SHARD_MESSAGE;
     header->payload_size = create_shard_message.size();
@@ -216,7 +216,8 @@ void HSHomeObject::on_shard_message_commit(int64_t lsn, sisl::blob const& header
     }
 
     auto shard = deserialize_shard(shard_msg);
-    if (shard.info.state == ShardInfo::State::OPEN) {
+    switch(shard.info.state) {
+    case ShardInfo::State::OPEN: {
         std::scoped_lock lock_guard(_flying_shard_lock);
         auto iter = _flying_shards.find(lsn);
         if (iter == _flying_shards.end()) {
@@ -228,21 +229,27 @@ void HSHomeObject::on_shard_message_commit(int64_t lsn, sisl::blob const& header
         }
         shard.chunk_id = iter->second.chunk_id;
         _flying_shards.erase(iter);
-        //deserialize the finalized shard msg;
+        //serialize the finalized shard msg;
         shard_msg = serialize_shard(shard);
-    }
-
-    // persist the serialize result to homestore MetaBlkService;
-    void* cookie = nullptr;
-    homestore::hs()->meta_service().add_sub_sb(HSHomeObject::s_shard_info_sub_type,
-                                               r_cast<const uint8_t*>(shard_msg.c_str()), shard_msg.size(), cookie);
-
-    //update in-memory shard map;
-    if (shard.info.state == ShardInfo::State::OPEN) {
-        //create shard;
+        // persist the serialize result to homestore MetaBlkService;
+        homestore::hs()->meta_service().add_sub_sb(HSHomeObject::s_shard_info_sub_type,
+            r_cast<const uint8_t*>(shard_msg.c_str()), shard_msg.size(), shard.metablk_cookie);
+        //update in-memory shard map;
         do_commit_new_shard(shard);
-    } else {
+        break;
+    }
+
+    case ShardInfo::State::SEALED: {
+        void* metablk_cookie = get_shard_metablk(shard.info.id);
+        RELEASE_ASSERT(metablk_cookie != nullptr, "seal shard when metablk is nullptr");
+        homestore::hs()->meta_service().update_sub_sb(r_cast<const uint8_t*>(shard_msg.c_str()), shard_msg.size(), metablk_cookie);
+        shard.metablk_cookie = metablk_cookie;
         do_commit_seal_shard(shard);
+        break;
+    }
+    default : {
+        break;
+    }
     }
 
     if (promise) {
@@ -273,4 +280,13 @@ void HSHomeObject::do_commit_seal_shard(const Shard& shard) {
     *(shard_iter->second) = shard;
 }
 
+void* HSHomeObject::get_shard_metablk(shard_id id) {
+    std::scoped_lock lock_guard(_shard_lock);
+    auto shard_iter = _shard_map.find(id);
+    if (shard_iter == _shard_map.end()) {
+        return nullptr;
+    }
+    return (*shard_iter->second).metablk_cookie;
+}
+
 } // namespace homeobject

src/lib/homestore/tests/test_shard_manager.cpp

@@ -8,6 +8,9 @@
 #include <sisl/logging/logging.h>
 #include <sisl/options/options.h>
 
+//will allow unit tests to access object private/protected for validation;
+#define protected public
+
 #include "lib/homestore/homeobject.hpp"
 #include "lib/homestore/replication_state_machine.hpp"
 #include "mocks/mock_replica_set.hpp"
@@ -134,6 +137,22 @@ TEST_F(ShardManagerWithShardsTesting, CreateShardSuccess) {
      EXPECT_EQ(_pg_id, shard_info.placement_group);    
 }
 
+TEST_F(ShardManagerWithShardsTesting, CreateShardAndValidateMembers) {
+     auto e = _home_object->shard_manager()->create_shard(_pg_id, Mi).get();
+     ASSERT_TRUE(!!e);
+     ShardInfo shard_info = e.value();
+     homeobject::HSHomeObject* ho = dynamic_cast<homeobject::HSHomeObject*>(_home_object.get());
+     EXPECT_TRUE(ho != nullptr);
+     auto pg_iter = ho->_pg_map.find(_pg_id);
+     EXPECT_TRUE(pg_iter !=  ho->_pg_map.end());
+     auto& pg = pg_iter->second;
+     EXPECT_TRUE(pg.shard_sequence_num == 1);
+     EXPECT_EQ(1, pg.shards.size());
+     auto& shard = *pg.shards.begin();
+     EXPECT_TRUE(shard.info == shard_info);
+     EXPECT_TRUE(shard.metablk_cookie != nullptr);
+}
+
 TEST_F(ShardManagerWithShardsTesting, GetKnownShard) {
      auto e = _home_object->shard_manager()->create_shard(_pg_id, Mi).get();
      ASSERT_TRUE(!!e);