diff --git a/conanfile.py b/conanfile.py
index fe81d8e7..f2702f34 100644
--- a/conanfile.py
+++ b/conanfile.py
@@ -9,7 +9,7 @@
 
 class HomeObjectConan(ConanFile):
     name = "homeobject"
-    version = "2.1.7"
+    version = "2.1.8"
 
     homepage = "https://github.com/eBay/HomeObject"
     description = "Blob Store built on HomeReplication"
diff --git a/src/include/homeobject/pg_manager.hpp b/src/include/homeobject/pg_manager.hpp
index 698ef9eb..4819052a 100644
--- a/src/include/homeobject/pg_manager.hpp
+++ b/src/include/homeobject/pg_manager.hpp
@@ -41,7 +41,8 @@ struct PGInfo {
     pg_id_t id;
     mutable MemberSet members;
     peer_id_t replica_set_uuid;
-    u_int64_t size;
+    uint64_t size;
+    uint64_t chunk_size;
 
     auto operator<=>(PGInfo const& rhs) const { return id <=> rhs.id; }
     auto operator==(PGInfo const& rhs) const { return id == rhs.id; }
diff --git a/src/include/homeobject/shard_manager.hpp b/src/include/homeobject/shard_manager.hpp
index 812ebe9f..19ee7e2b 100644
--- a/src/include/homeobject/shard_manager.hpp
+++ b/src/include/homeobject/shard_manager.hpp
@@ -10,7 +10,7 @@
 namespace homeobject {
 
 ENUM(ShardError, uint16_t, UNKNOWN = 1, TIMEOUT, INVALID_ARG, NOT_LEADER, UNSUPPORTED_OP, UNKNOWN_PG, UNKNOWN_SHARD,
-     PG_NOT_READY, CRC_MISMATCH);
+     PG_NOT_READY, CRC_MISMATCH, NO_SPACE_LEFT);
 
 struct ShardInfo {
     enum class State : uint8_t {
diff --git a/src/lib/homestore_backend/heap_chunk_selector.cpp b/src/lib/homestore_backend/heap_chunk_selector.cpp
index 6e2a1492..f18b7924 100644
--- a/src/lib/homestore_backend/heap_chunk_selector.cpp
+++ b/src/lib/homestore_backend/heap_chunk_selector.cpp
@@ -20,11 +20,7 @@ namespace homeobject {
 void HeapChunkSelector::add_chunk(csharedChunk& chunk) { m_chunks.emplace(VChunk(chunk).get_chunk_id(), chunk); }
 
 void HeapChunkSelector::add_chunk_internal(const chunk_num_t chunkID, bool add_to_heap) {
-    if (m_chunks.find(chunkID) == m_chunks.end()) {
-        // sanity check
-        LOGWARNMOD(homeobject, "No chunk found for ChunkID {}", chunkID);
-        return;
-    }
+    // private function chunkID must belong to m_chunks
 
     const auto& chunk = m_chunks[chunkID];
     VChunk vchunk(chunk);
@@ -59,39 +55,21 @@ csharedChunk HeapChunkSelector::select_chunk(homestore::blk_count_t count, const
         return nullptr;
     }
 
-    std::shared_lock lock_guard(m_chunk_selector_mtx);
-    // FIXME @Hooper: Temporary bypass using pdev_id_hint to represent pg_id_hint, "identical layout" will change it
     pg_id_t pg_id = 0;
-    auto& pg_id_hint = hint.pdev_id_hint;
-    if (!pg_id_hint.has_value()) {
+    homestore::chunk_num_t chunk_id = 0;
+    if (!hint.pdev_id_hint.has_value()) {
         LOGWARNMOD(homeobject, "should not allocated a chunk without exiting pg_id in hint!");
         return nullptr;
     } else {
-        pg_id = pg_id_hint.value();
-    }
-
-    auto it = m_per_pg_heap.find(pg_id);
-    if (it == m_per_pg_heap.end()) {
-        LOGWARNMOD(homeobject, "No pg found for pg_id {}", pg_id);
-        return nullptr;
+        // use pdev_id_hint (of type uint32_t) to store the values of pg_id and chunk_id.
+        // Both chunk_num_t and pg_id_t are of type uint16_t.
+        static_assert(std::is_same<pg_id_t, uint16_t>::value, "pg_id_t is not uint16_t");
+        static_assert(std::is_same<homestore::chunk_num_t, uint16_t>::value, "chunk_num_t is not uint16_t");
+        uint32_t pdev_id_hint = hint.pdev_id_hint.value();
+        pg_id = (uint16_t)(pdev_id_hint >> 16);
+        chunk_id = (uint16_t)(pdev_id_hint & 0xFFFF);
+        return select_specific_chunk(pg_id, chunk_id);
     }
-
-    auto vchunk = VChunk(nullptr);
-    auto& heap = it->second->m_heap;
-    if (auto lock_guard = std::lock_guard< std::mutex >(it->second->mtx); !heap.empty()) {
-        vchunk = heap.top();
-        heap.pop();
-    }
-
-    if (vchunk.get_internal_chunk()) {
-        auto& avalableBlkCounter = it->second->available_blk_count;
-        avalableBlkCounter.fetch_sub(vchunk.available_blks());
-        remove_chunk_from_defrag_heap(vchunk.get_chunk_id());
-    } else {
-        LOGWARNMOD(homeobject, "no available chunks left for pg {}", pg_id);
-    }
-
-    return vchunk.get_internal_chunk();
 }
 
 csharedChunk HeapChunkSelector::select_specific_chunk(const pg_id_t pg_id, const chunk_num_t chunkID) {
@@ -131,6 +109,8 @@ csharedChunk HeapChunkSelector::select_specific_chunk(const pg_id_t pg_id, const
         auto& avalableBlkCounter = pg_it->second->available_blk_count;
         avalableBlkCounter.fetch_sub(vchunk.available_blks());
         remove_chunk_from_defrag_heap(vchunk.get_chunk_id());
+    } else {
+        LOGWARNMOD(homeobject, "No chunk found for ChunkID {} in PG {}", chunkID, pg_id);
     }
 
     return vchunk.get_internal_chunk();
@@ -175,18 +155,18 @@ void HeapChunkSelector::foreach_chunks(std::function< void(csharedChunk&) >&& cb
                   [cb = std::move(cb)](auto& p) { cb(p.second); });
 }
 
-void HeapChunkSelector::release_chunk(const pg_id_t pg_id, const chunk_num_t chunkID) {
+bool HeapChunkSelector::release_chunk(const pg_id_t pg_id, const chunk_num_t chunkID) {
     std::shared_lock lock_guard(m_chunk_selector_mtx);
     if (m_chunks.find(chunkID) == m_chunks.end()) {
         // sanity check
         LOGWARNMOD(homeobject, "No chunk found for ChunkID {}", chunkID);
-        return;
+        return false;
     }
 
     auto pg_it = m_per_pg_heap.find(pg_id);
     if (pg_it == m_per_pg_heap.end()) {
         LOGWARNMOD(homeobject, "No pg found for pg_id {}", pg_id);
-        return;
+        return false;
     }
 
     const auto& chunk = m_chunks[chunkID];
@@ -199,6 +179,7 @@ void HeapChunkSelector::release_chunk(const pg_id_t pg_id, const chunk_num_t chu
     auto& avalableBlkCounter = pg_it->second->available_blk_count;
     avalableBlkCounter += vchunk.available_blks();
 
+    return true;
 }
 
 uint32_t HeapChunkSelector::get_chunk_size() const {
@@ -231,12 +212,12 @@ std::optional< uint32_t > HeapChunkSelector::select_chunks_for_pg(pg_id_t pg_id,
     }
     auto vchunk = VChunk(nullptr);
     auto it = m_per_pg_heap.emplace(pg_id, std::make_shared< ChunkHeap >()).first;
-    auto v2r_vector = m_v2r_chunk_map.emplace(pg_id, std::make_shared< std::vector < chunk_num_t > >()).first->second;
-    auto r2v_map = m_r2v_chunk_map.emplace(pg_id, std::make_shared< ChunkIdMap >()).first->second;
+    auto v2p_vector = m_v2p_chunk_map.emplace(pg_id, std::make_shared< std::vector < chunk_num_t > >()).first->second;
+    auto p2v_map = m_p2v_chunk_map.emplace(pg_id, std::make_shared< ChunkIdMap >()).first->second;
 
     auto& pg_heap = it->second;
     std::scoped_lock lock(pdev_heap->mtx, pg_heap->mtx);
-    v2r_vector->reserve(num_chunk);
+    v2p_vector->reserve(num_chunk);
     for (chunk_num_t i = 0; i < num_chunk; ++i) {
         vchunk = pdev_heap->m_heap.top();
         //sanity check
@@ -249,58 +230,76 @@ std::optional< uint32_t > HeapChunkSelector::select_chunks_for_pg(pg_id_t pg_id,
         pg_heap->available_blk_count += vchunk.available_blks();
         // v_chunk_id start from 0.
         chunk_num_t v_chunk_id = i;
-        chunk_num_t r_chunk_id = vchunk.get_chunk_id();
-        v2r_vector->emplace_back(r_chunk_id);
-        r2v_map->emplace(r_chunk_id, v_chunk_id);
+        chunk_num_t p_chunk_id = vchunk.get_chunk_id();
+        v2p_vector->emplace_back(p_chunk_id);
+        p2v_map->emplace(p_chunk_id, v_chunk_id);
     }
 
     return num_chunk;
 }
 
-void HeapChunkSelector::set_pg_chunks(pg_id_t pg_id, std::vector<chunk_num_t>&& chunk_ids) {
+bool HeapChunkSelector::set_pg_chunks(pg_id_t pg_id, std::vector<chunk_num_t>&& chunk_ids) {
     std::unique_lock lock_guard(m_chunk_selector_mtx);
-    if (m_v2r_chunk_map.find(pg_id) != m_v2r_chunk_map.end()) {
+    // check pg exist
+    if (m_v2p_chunk_map.find(pg_id) != m_v2p_chunk_map.end()) {
         LOGWARNMOD(homeobject, "PG {} had been recovered", pg_id);
-        return;
+        return false;
+    }
+
+    // check chunks valid, must belong to m_chunks and have same pdev_id
+    std::optional<uint32_t> last_pdev_id;
+    for (auto chunk_id : chunk_ids) {
+        auto it = m_chunks.find(chunk_id);
+        if (it == m_chunks.end()) {
+            LOGWARNMOD(homeobject, "No chunk found for ChunkID {}", chunk_id);
+            return false;
+        }
+        VChunk v_chunk(it->second);
+        if (last_pdev_id.has_value() && last_pdev_id.value() != v_chunk.get_pdev_id()) {
+            LOGWARNMOD(homeobject, "The pdev value is different, last_pdev_id={}, pdev_id={}", last_pdev_id.value(), v_chunk.get_pdev_id());
+            return false;
+        } else {
+            last_pdev_id = v_chunk.get_pdev_id();
+        }
     }
 
-    auto v2r_vector = m_v2r_chunk_map.emplace(pg_id, std::make_shared< std::vector < chunk_num_t > >(std::move(chunk_ids))).first->second;
-    auto r2v_map = m_r2v_chunk_map.emplace(pg_id, std::make_shared< ChunkIdMap >()).first->second;
+    auto v2p_vector = m_v2p_chunk_map.emplace(pg_id, std::make_shared< std::vector < chunk_num_t > >(std::move(chunk_ids))).first->second;
+    auto p2v_map = m_p2v_chunk_map.emplace(pg_id, std::make_shared< ChunkIdMap >()).first->second;
 
-    for (chunk_num_t i = 0; i < v2r_vector->size(); ++i) {
+    for (chunk_num_t i = 0; i < v2p_vector->size(); ++i) {
         // v_chunk_id start from 0.
         chunk_num_t v_chunk_id = i;
-        chunk_num_t r_chunk_id = (*v2r_vector)[i];
-        r2v_map->emplace(r_chunk_id, v_chunk_id);
+        chunk_num_t p_chunk_id = (*v2p_vector)[i];
+        p2v_map->emplace(p_chunk_id, v_chunk_id);
     }
+    return true;
 }
 
 void HeapChunkSelector::recover_per_dev_chunk_heap() {
     std::unique_lock lock_guard(m_chunk_selector_mtx);
     for (const auto& [chunk_id, _] : m_chunks) {
         bool add_to_heap = true;
-        for (const auto& [_, chunk_map] : m_r2v_chunk_map) {
+        for (const auto& [_, chunk_map] : m_p2v_chunk_map) {
             if (chunk_map->find(chunk_id) != chunk_map->end()) {
                 add_to_heap = false;
                 break;
             }
         }
         add_chunk_internal(chunk_id, add_to_heap);
-
     }
 }
 
-void HeapChunkSelector::recover_pg_chunk_heap(pg_id_t pg_id, const std::unordered_set< chunk_num_t >& excludingChunks)
+bool HeapChunkSelector::recover_pg_chunk_heap(pg_id_t pg_id, const std::unordered_set< chunk_num_t >& excludingChunks)
 {
     std::unique_lock lock_guard(m_chunk_selector_mtx);
     if (m_per_pg_heap.find(pg_id) != m_per_pg_heap.end()) {
         LOGWARNMOD(homeobject, "Pg_heap {} had been recovered", pg_id);
-        return;
+        return false;
     }
-    auto it = m_v2r_chunk_map.find(pg_id);
-    if (it == m_v2r_chunk_map.end()) {
-        LOGWARNMOD(homeobject, "Pg_chunk_map {} had never been recovered", pg_id);
-        return;
+    auto it = m_v2p_chunk_map.find(pg_id);
+    if (it == m_v2p_chunk_map.end()) {
+        LOGWARNMOD(homeobject, "Pg_chunk_map {} should be recovered beforehand", pg_id);
+        return false;
     }
     const auto& chunk_ids = it->second;
     auto& pg_heap = m_per_pg_heap.emplace(pg_id, std::make_shared< ChunkHeap >()).first->second;
@@ -313,12 +312,13 @@ void HeapChunkSelector::recover_pg_chunk_heap(pg_id_t pg_id, const std::unordere
             pg_heap->available_blk_count += vchunk.available_blks();
         }
     }
+    return true;
 }
 
 std::shared_ptr< const std::vector <homestore::chunk_num_t> > HeapChunkSelector::get_pg_chunks(pg_id_t pg_id) const {
     std::shared_lock lock_guard(m_chunk_selector_mtx);
-    auto it = m_v2r_chunk_map.find(pg_id);
-    if (it != m_v2r_chunk_map.end()) {
+    auto it = m_v2p_chunk_map.find(pg_id);
+    if (it != m_v2p_chunk_map.end()) {
         return it->second;
     } else {
         LOGWARNMOD(homeobject, "PG {} had never been created", pg_id);
@@ -326,6 +326,57 @@ std::shared_ptr< const std::vector <homestore::chunk_num_t> > HeapChunkSelector:
     }
 }
 
+std::optional< homestore::chunk_num_t > HeapChunkSelector::peek_top_chunk(pg_id_t pg_id) const {
+    std::shared_lock lock_guard(m_chunk_selector_mtx);
+    auto pg_it = m_per_pg_heap.find(pg_id);
+    if (pg_it == m_per_pg_heap.end()) {
+        LOGWARNMOD(homeobject, "No pg found for pg_id {}", pg_id);
+        return std::nullopt;
+    }
+    VChunk vchunk(nullptr);
+    auto& heap = pg_it->second->m_heap;
+    if (auto lock_guard = std::lock_guard< std::mutex >(pg_it->second->mtx); !heap.empty()) {
+        vchunk = heap.top();
+    }
+
+    if (vchunk.get_internal_chunk()) {
+        return vchunk.get_chunk_id();
+    } else {
+        return std::nullopt;
+    }
+}
+
+std::optional< homestore::chunk_num_t > HeapChunkSelector::get_virtual_chunk_id(pg_id_t pg_id, chunk_num_t p_chunk_id) const {
+    std::shared_lock lock_guard(m_chunk_selector_mtx);
+    auto pg_it = m_p2v_chunk_map.find(pg_id);
+    if (pg_it == m_p2v_chunk_map.end()) {
+        LOGWARNMOD(homeobject, "No pg found for pg_id {}", pg_id);
+        return std::nullopt;
+    }
+    auto& p2v_map = pg_it->second;
+    auto it = p2v_map->find(p_chunk_id);
+    if (it == p2v_map->end()) {
+        return std::nullopt;
+    } else {
+        return it->second;
+    }
+}
+
+std::optional< homestore::chunk_num_t > HeapChunkSelector::get_physical_chunk_id(pg_id_t pg_id, chunk_num_t v_chunk_id) const {
+    std::shared_lock lock_guard(m_chunk_selector_mtx);
+    auto pg_it = m_v2p_chunk_map.find(pg_id);
+    if (pg_it == m_v2p_chunk_map.end()) {
+        LOGWARNMOD(homeobject, "No pg found for pg_id {}", pg_id);
+        return std::nullopt;
+    }
+    auto& v2p_vec = pg_it->second;
+    if (v_chunk_id >= v2p_vec->size()) {
+        return std::nullopt;
+    } else {
+        return (*v2p_vec)[v_chunk_id];
+    }
+}
+
 homestore::blk_alloc_hints HeapChunkSelector::chunk_to_hints(chunk_num_t chunk_id) const {
     auto iter = m_chunks.find(chunk_id);
     if (iter == m_chunks.end()) {
diff --git a/src/lib/homestore_backend/heap_chunk_selector.h b/src/lib/homestore_backend/heap_chunk_selector.h
index 259ecfb5..705861b3 100644
--- a/src/lib/homestore_backend/heap_chunk_selector.h
+++ b/src/lib/homestore_backend/heap_chunk_selector.h
@@ -35,7 +35,7 @@ class HeapChunkSelector : public homestore::ChunkSelector {
 
     using VChunkHeap = std::priority_queue< VChunk, std::vector< VChunk >, VChunkComparator >;
     using VChunkDefragHeap = std::priority_queue< VChunk, std::vector< VChunk >, VChunkDefragComparator >;
-    using ChunkIdMap = std::unordered_map < homestore::chunk_num_t, homestore::chunk_num_t >; // used for real chunk id -> virtual chunk id map
+    using ChunkIdMap = std::unordered_map < homestore::chunk_num_t, homestore::chunk_num_t >; // used for physical chunk id -> virtual chunk id map
     using chunk_num_t = homestore::chunk_num_t;
 
     struct ChunkHeap {
@@ -52,7 +52,7 @@ class HeapChunkSelector : public homestore::ChunkSelector {
 
     csharedChunk select_chunk([[maybe_unused]] homestore::blk_count_t nblks, const homestore::blk_alloc_hints& hints);
 
-    // this function will be used by GC flow or recovery flow to mark one specific chunk to be busy, caller should be
+    // this function will be used by create shard or recovery flow to mark one specific chunk to be busy, caller should be
     // responsible to use release_chunk() interface to release it when no longer to use the chunk anymore.
     csharedChunk select_specific_chunk(const pg_id_t pg_id, const chunk_num_t);
 
@@ -61,7 +61,7 @@ class HeapChunkSelector : public homestore::ChunkSelector {
 
     // this function is used to return a chunk back to ChunkSelector when sealing a shard, and will only be used by
     // Homeobject.
-    void release_chunk(const pg_id_t pg_id, const chunk_num_t);
+    bool release_chunk(const pg_id_t pg_id, const chunk_num_t);
 
     /**
      * select chunks for pg, chunks need to be in same pdev.
@@ -74,14 +74,20 @@ class HeapChunkSelector : public homestore::ChunkSelector {
 
     std::shared_ptr< const std::vector <chunk_num_t> > get_pg_chunks(pg_id_t pg_id) const;
 
+    std::optional< chunk_num_t > peek_top_chunk(pg_id_t pg_id) const;
+
+    std::optional< chunk_num_t > get_virtual_chunk_id(pg_id_t pg_id, chunk_num_t p_chunk_id) const;
+
+    std::optional< chunk_num_t > get_physical_chunk_id(pg_id_t pg_id, chunk_num_t v_chunk_id) const;
+
     // this should be called on each pg meta blk found
-    void set_pg_chunks(pg_id_t pg_id, std::vector<chunk_num_t>&& chunk_ids);
+    bool set_pg_chunks(pg_id_t pg_id, std::vector<chunk_num_t>&& chunk_ids);
 
     // this should be called after all pg meta blk recovered
     void recover_per_dev_chunk_heap();
 
     // this should be called after ShardManager is initialized and get all the open shards
-    void recover_pg_chunk_heap(pg_id_t pg_id, const std::unordered_set< chunk_num_t >& excludingChunks);
+    bool recover_pg_chunk_heap(pg_id_t pg_id, const std::unordered_set< chunk_num_t >& excludingChunks);
 
     /**
      * Retrieves the block allocation hints for a given chunk.
@@ -138,11 +144,11 @@ class HeapChunkSelector : public homestore::ChunkSelector {
     std::unordered_map< uint32_t, std::shared_ptr< ChunkHeap > > m_per_dev_heap;
     std::unordered_map< pg_id_t, std::shared_ptr< ChunkHeap > > m_per_pg_heap;
 
-    // These mappings ensure "identical layout" by providing bidirectional indexing between virtual and real chunk IDs.
-    // m_v2r_chunk_map: Maps each pg_id to a vector of real chunk IDs (r_chunk_id). The index in the vector corresponds to the virtual chunk ID (v_chunk_id).
-    std::unordered_map< pg_id_t, std::shared_ptr< std::vector <chunk_num_t> > > m_v2r_chunk_map;
-    // m_r2v_chunk_map: Maps each pg_id to a map that inversely maps real chunk IDs (r_chunk_id) to virtual chunk IDs (v_chunk_id).
-    std::unordered_map< pg_id_t, std::shared_ptr< ChunkIdMap > > m_r2v_chunk_map;
+    // These mappings ensure "identical layout" by providing bidirectional indexing between virtual and physical chunk IDs.
+    // m_v2p_chunk_map: Maps each pg_id to a vector of physical chunk IDs (p_chunk_id). The index in the vector corresponds to the virtual chunk ID (v_chunk_id).
+    std::unordered_map< pg_id_t, std::shared_ptr< std::vector <chunk_num_t> > > m_v2p_chunk_map;
+    // m_p2v_chunk_map: Maps each pg_id to a map that inversely maps physical chunk IDs (p_chunk_id) to virtual chunk IDs (v_chunk_id).
+    std::unordered_map< pg_id_t, std::shared_ptr< ChunkIdMap > > m_p2v_chunk_map;
 
     // hold all the chunks , selected or not
     std::unordered_map< chunk_num_t, csharedChunk > m_chunks;
diff --git a/src/lib/homestore_backend/hs_homeobject.hpp b/src/lib/homestore_backend/hs_homeobject.hpp
index d2a46892..7c6dd970 100644
--- a/src/lib/homestore_backend/hs_homeobject.hpp
+++ b/src/lib/homestore_backend/hs_homeobject.hpp
@@ -90,9 +90,9 @@ class HSHomeObject : public HomeObjectImpl {
         // Data layout inside 'data':
         // First, an array of 'pg_members' structures:
         // | pg_members[0] | pg_members[1] | ... | pg_members[num_members-1] |
-        // Immediately followed by an array of 'chunk_num_t' values (representing r_chunk_ids):
+        // Immediately followed by an array of 'chunk_num_t' values (representing physical chunkID):
         // | chunk_num_t[0] | chunk_num_t[1] | ... | chunk_num_t[num_chunks-1] |
-        // Here, 'chunk_num_t[i]' represents the r_chunk_id for the v_chunk_id 'i', where v_chunk_id starts from 0 and increases sequentially.
+        // Here, 'chunk_num_t[i]' represents the p_chunk_id for the v_chunk_id 'i', where v_chunk_id starts from 0 and increases sequentially.
 
 
         uint32_t size() const { return sizeof(pg_info_superblk)  - sizeof(char) + num_members * sizeof(pg_members) + num_chunks * sizeof(homestore::chunk_num_t); }
@@ -142,6 +142,7 @@ class HSHomeObject : public HomeObjectImpl {
     struct shard_info_superblk : public DataHeader {
         ShardInfo info;
         homestore::chunk_num_t chunk_id;
+        homestore::chunk_num_t v_chunk_id;
     };
 #pragma pack()
 
@@ -210,7 +211,6 @@ class HSHomeObject : public HomeObjectImpl {
     public:
         homestore::superblk< pg_info_superblk > pg_sb_;
         shared< homestore::ReplDev > repl_dev_;
-        std::optional< homestore::chunk_num_t > any_allocated_chunk_id_{};
         std::shared_ptr< BlobIndexTable > index_table_;
         PGMetrics metrics_;
 
@@ -440,12 +440,12 @@ class HSHomeObject : public HomeObjectImpl {
     void on_replica_restart();
 
     /**
-     * @brief Returns any chunk number for the given pg ID.
+     * @brief Extracts the physical chunk ID for create shard from the message.
      *
-     * @param pg The pg ID to get the chunk number for.
-     * @return A tuple of <if pg exist, if shard exist, chunk number if both exist>.
+     * @param header The message header that includes the shard_info_superblk, which contains the data necessary for extracting and mapping the chunk ID.
+     * @return An optional pyhsical chunk id if the extraction and mapping process is successful, otherwise an empty optional.
      */
-    std::tuple< bool, bool, homestore::chunk_num_t > get_any_chunk_id(pg_id_t pg);
+    std::optional< homestore::chunk_num_t > resolve_physical_chunk_id_from_msg(sisl::blob const& header);
 
     cshared< HeapChunkSelector > chunk_selector() const { return chunk_selector_; }
 
diff --git a/src/lib/homestore_backend/hs_pg_manager.cpp b/src/lib/homestore_backend/hs_pg_manager.cpp
index 0cd9f749..3891782a 100644
--- a/src/lib/homestore_backend/hs_pg_manager.cpp
+++ b/src/lib/homestore_backend/hs_pg_manager.cpp
@@ -69,6 +69,7 @@ PGManager::NullAsyncResult HSHomeObject::_create_pg(PGInfo&& pg_info, std::set<
         return folly::makeUnexpected(PGError::NO_SPACE_LEFT);
     }
 
+    pg_info.chunk_size = chunk_size;
     pg_info.replica_set_uuid = boost::uuids::random_generator()();
     return hs_repl_service()
         .create_repl_dev(pg_info.replica_set_uuid, peers)
@@ -136,15 +137,26 @@ void HSHomeObject::on_create_pg_message_commit(int64_t lsn, sisl::blob const& he
         return;
     }
 
+    auto local_chunk_size = chunk_selector()->get_chunk_size();
+    if (pg_info.chunk_size != local_chunk_size) {
+        LOGE("Chunk sizes are inconsistent, leader_chunk_size={}, local_chunk_size={}", pg_info.chunk_size, local_chunk_size);
+        if (ctx) { ctx->promise_.setValue(folly::makeUnexpected(PGError::UNKNOWN)); }
+        return;
+    }
+
     // select chunks for pg
     auto const num_chunk = chunk_selector()->select_chunks_for_pg(pg_id, pg_info.size);
     if (!num_chunk.has_value()) {
-        LOGW("select chunks for pg failed, pg_id {}", pg_id);
+        LOGW("Failed to select chunks for pg {}", pg_id);
         if (ctx) { ctx->promise_.setValue(folly::makeUnexpected(PGError::NO_SPACE_LEFT)); }
         return;
     }
     auto chunk_ids = chunk_selector()->get_pg_chunks(pg_id);
-
+    if (chunk_ids == nullptr) {
+        LOGW("Failed to get pg chunks, pg_id {}", pg_id);
+        if (ctx) { ctx->promise_.setValue(folly::makeUnexpected(PGError::NO_SPACE_LEFT)); }
+        return; 
+    }
     // create index table and pg
     // TODO create index table during create shard.
     auto index_table = create_index_table();
@@ -245,6 +257,7 @@ std::string HSHomeObject::serialize_pg_info(const PGInfo& pginfo) {
     nlohmann::json j;
     j["pg_info"]["pg_id_t"] = pginfo.id;
     j["pg_info"]["pg_size"] = pginfo.size;
+    j["pg_info"]["chunk_size"] = pginfo.chunk_size;
     j["pg_info"]["repl_uuid"] = boost::uuids::to_string(pginfo.replica_set_uuid);
 
     nlohmann::json members_j{};
@@ -263,7 +276,8 @@ PGInfo HSHomeObject::deserialize_pg_info(const unsigned char* json_str, size_t s
     auto pg_json = nlohmann::json::parse(json_str, json_str + size);
 
     PGInfo pg_info(pg_json["pg_info"]["pg_id_t"].get< pg_id_t >());
-    pg_info.size = pg_json["pg_info"]["pg_size"].get< u_int64_t >();
+    pg_info.size = pg_json["pg_info"]["pg_size"].get< uint64_t >();
+    pg_info.chunk_size = pg_json["pg_info"]["chunk_size"].get< uint64_t >();
     pg_info.replica_set_uuid = boost::uuids::string_generator()(pg_json["pg_info"]["repl_uuid"].get< std::string >());
 
     for (auto const& m : pg_json["pg_info"]["members"]) {
@@ -288,7 +302,8 @@ void HSHomeObject::on_pg_meta_blk_found(sisl::byte_view const& buf, void* meta_c
     }
     auto pg_id = pg_sb->id;
     std::vector<chunk_num_t> chunk_ids(pg_sb->get_chunk_ids(), pg_sb->get_chunk_ids() + pg_sb->num_chunks);
-    chunk_selector_->set_pg_chunks(pg_id, std::move(chunk_ids));
+    bool set_pg_chunks_res = chunk_selector_->set_pg_chunks(pg_id, std::move(chunk_ids));
+    RELEASE_ASSERT(set_pg_chunks_res, "Failed to set pg={} chunks", pg_id);
     auto uuid_str = boost::uuids::to_string(pg_sb->index_table_uuid);
     auto hs_pg = std::make_unique< HS_PG >(std::move(pg_sb), std::move(v.value()));
     // During PG recovery check if index is already recoverd else
diff --git a/src/lib/homestore_backend/hs_shard_manager.cpp b/src/lib/homestore_backend/hs_shard_manager.cpp
index 938c64ec..9c111313 100644
--- a/src/lib/homestore_backend/hs_shard_manager.cpp
+++ b/src/lib/homestore_backend/hs_shard_manager.cpp
@@ -111,6 +111,16 @@ ShardManager::AsyncResult< ShardInfo > HSHomeObject::_create_shard(pg_id_t pg_ow
     auto new_shard_id = generate_new_shard_id(pg_owner);
     auto create_time = get_current_timestamp();
 
+    // select chunk for shard.
+    const auto p_chunkID = chunk_selector()->peek_top_chunk(pg_owner);
+    if (!p_chunkID.has_value()) {
+        LOGW("no availble chunk left to create shard for pg [{}]", pg_owner);
+        return folly::makeUnexpected(ShardError::NO_SPACE_LEFT);
+    }
+    const auto v_chunkID = chunk_selector()->get_virtual_chunk_id(pg_owner, p_chunkID.value());
+    RELEASE_ASSERT(v_chunkID.has_value(), "No virutal chunk id for physical chunk id {} in pg {}", p_chunkID.value(),
+                   pg_owner);
+
     // Prepare the shard info block
     sisl::io_blob_safe sb_blob(sisl::round_up(sizeof(shard_info_superblk), repl_dev->get_blk_size()), io_align);
     shard_info_superblk* sb = new (sb_blob.bytes()) shard_info_superblk();
@@ -125,6 +135,7 @@ ShardManager::AsyncResult< ShardInfo > HSHomeObject::_create_shard(pg_id_t pg_ow
                          .total_capacity_bytes = size_bytes,
                          .deleted_capacity_bytes = 0};
     sb->chunk_id = 0;
+    sb->v_chunk_id = v_chunkID.value();
 
     auto req = repl_result_ctx< ShardManager::Result< ShardInfo > >::make(
         sizeof(shard_info_superblk) /* header_extn_size */, 0u /* key_size */);
@@ -365,7 +376,8 @@ void HSHomeObject::on_shard_message_commit(int64_t lsn, sisl::blob const& h, hom
             auto pg_id = shard_info.placement_group;
             auto chunk_id = get_shard_chunk(shard_info.id);
             RELEASE_ASSERT(chunk_id.has_value(), "Chunk id not found");
-            chunk_selector()->release_chunk(pg_id, chunk_id.value());
+            bool res = chunk_selector()->release_chunk(pg_id, chunk_id.value());
+            RELEASE_ASSERT(res, "Failed to release chunk {}, pg_id {}", chunk_id.value(), pg_id);
             update_shard_in_map(shard_info);
         } else
             LOGW("try to commit SEAL_SHARD_MSG but shard state is not sealed, shard_id: {}", shard_info.id);
@@ -395,7 +407,8 @@ void HSHomeObject::on_shard_meta_blk_recover_completed(bool success) {
                 excluding_chunks.emplace(d_cast< HS_Shard* >(shard.get())->sb_->chunk_id);
             }
         }
-        chunk_selector_->recover_pg_chunk_heap(pair.first, excluding_chunks);
+        bool res = chunk_selector_->recover_pg_chunk_heap(pair.first, excluding_chunks);
+        RELEASE_ASSERT(res, "Failed to recover pg chunk heap, pg={}", pair.first);
     }
 }
 
@@ -432,22 +445,38 @@ std::optional< homestore::chunk_num_t > HSHomeObject::get_shard_chunk(shard_id_t
     return std::make_optional< homestore::chunk_num_t >(hs_shard->sb_->chunk_id);
 }
 
-std::tuple< bool, bool, homestore::chunk_num_t > HSHomeObject::get_any_chunk_id(pg_id_t pg_id) {
-    std::scoped_lock lock_guard(_pg_lock);
-    auto pg_iter = _pg_map.find(pg_id);
-    if (pg_iter == _pg_map.end()) { return {false /* pg_found */, false /* shards_found */, 0 /* chunk_id */}; }
-
-    HS_PG* pg = static_cast< HS_PG* >(pg_iter->second.get());
-    if (pg->any_allocated_chunk_id_.has_value()) { // it is already cached and use it;
-        return {true /* pg_found */, true /* shards_found */, *pg->any_allocated_chunk_id_};
+std::optional< homestore::chunk_num_t > HSHomeObject::resolve_physical_chunk_id_from_msg(sisl::blob const& header) {
+    const ReplicationMessageHeader* msg_header = r_cast< const ReplicationMessageHeader* >(header.cbytes());
+    if (msg_header->corrupted()) {
+        LOGW("replication message header is corrupted with crc error");
+        return std::nullopt;
     }
 
-    auto& shards = pg->shards_;
-    if (shards.empty()) { return {true /* pg_found */, false /* shards_found */, 0 /* chunk_id */}; }
-
-    auto hs_shard = d_cast< HS_Shard* >(shards.front().get());
-    pg->any_allocated_chunk_id_ = hs_shard->sb_->chunk_id; // cache it;
-    return {true /* pg_found */, true /* shards_found */, *pg->any_allocated_chunk_id_};
+    switch (msg_header->msg_type) {
+    case ReplicationMessageType::CREATE_SHARD_MSG: {
+        const pg_id_t pg_id = msg_header->pg_id;
+        std::scoped_lock lock_guard(_pg_lock);
+        auto pg_iter = _pg_map.find(pg_id);
+        if (pg_iter == _pg_map.end()) {
+            LOGW("Requesting a chunk for an unknown pg={}", pg_id);
+            return std::nullopt;
+        }
+        auto sb = r_cast< shard_info_superblk const* >(header.cbytes() + sizeof(ReplicationMessageHeader));
+        auto chunk_id = chunk_selector()->get_physical_chunk_id(pg_id, sb->v_chunk_id);
+        if (!chunk_id.has_value()) {
+            LOGW("Failed get physical chunk id by virtual chunk id, pg {}, v_chunk_id {}", pg_id, sb->v_chunk_id);
+            return std::nullopt;
+        } else {
+            return chunk_id;
+        }
+        break;
+    }
+    default: {
+        LOGW("Unexpected message type encountered: {}. This function should only be called with 'CREATE_SHARD_MSG'.",
+             msg_header->msg_type);
+        return std::nullopt;
+    }
+    }
 }
 
 HSHomeObject::HS_Shard::HS_Shard(ShardInfo shard_info, homestore::chunk_num_t chunk_id) :
diff --git a/src/lib/homestore_backend/replication_state_machine.cpp b/src/lib/homestore_backend/replication_state_machine.cpp
index ac3c6114..f5894664 100644
--- a/src/lib/homestore_backend/replication_state_machine.cpp
+++ b/src/lib/homestore_backend/replication_state_machine.cpp
@@ -121,9 +121,16 @@ ReplicationStateMachine::get_blk_alloc_hints(sisl::blob const& header, uint32_t
     const ReplicationMessageHeader* msg_header = r_cast< const ReplicationMessageHeader* >(header.cbytes());
     switch (msg_header->msg_type) {
     case ReplicationMessageType::CREATE_SHARD_MSG: {
-        // Since chunks are selected when a pg is created, the chunkselector selects one of the chunks owned by the pg
+        auto p_chunkID = home_object_->resolve_physical_chunk_id_from_msg(header);
+        RELEASE_ASSERT(p_chunkID.has_value(), "unkown chunk for create shard");
         homestore::blk_alloc_hints hints;
-        hints.pdev_id_hint = msg_header->pg_id; // FIXME @Hooper: Temporary bypass using pdev_id_hint to represent pg_id_hint, "identical layout" will change it
+        // use pdev_id_hint (of type uint32_t) to store the values of pg_id and chunk_id.
+        // Both chunk_num_t and pg_id_t are of type uint16_t.
+        static_assert(std::is_same<pg_id_t, uint16_t>::value, "pg_id_t is not uint16_t");
+        static_assert(std::is_same<homestore::chunk_num_t, uint16_t>::value, "chunk_num_t is not uint16_t");
+        homestore::chunk_num_t p_chunk_id = p_chunkID.value();
+        pg_id_t pg_id = msg_header->pg_id;
+        hints.pdev_id_hint = ((uint32_t)pg_id << 16) | p_chunk_id;
         return hints;
     }
 
diff --git a/src/lib/homestore_backend/tests/hs_pg_tests.cpp b/src/lib/homestore_backend/tests/hs_pg_tests.cpp
index b8a4499f..fd7ddf79 100644
--- a/src/lib/homestore_backend/tests/hs_pg_tests.cpp
+++ b/src/lib/homestore_backend/tests/hs_pg_tests.cpp
@@ -46,6 +46,47 @@ TEST_F(HomeObjectFixture, PGStatsTest) {
     LOGINFO("HomeObj stats: {}", stats.to_string());
 }
 
+
+TEST_F(HomeObjectFixture, PGExceedSpaceTest) {
+    LOGINFO("HomeObject replica={} setup completed", g_helper->replica_num());
+    pg_id_t pg_id{1};
+    if (0 == g_helper->replica_num()) { // leader
+        auto memebers = g_helper->members();
+        auto name = g_helper->name();
+        auto info = homeobject::PGInfo(pg_id);
+        info.size = 500 * Gi; // execced local available space
+        for (const auto& member : memebers) {
+            if (0 == member.second) {
+                // by default, leader is the first member
+                info.members.insert(homeobject::PGMember{member.first, name + std::to_string(member.second), 1});
+            } else {
+                info.members.insert(homeobject::PGMember{member.first, name + std::to_string(member.second), 0});
+            }
+        }
+        auto p = _obj_inst->pg_manager()->create_pg(std::move(info)).get();
+        
+        ASSERT_TRUE(p.hasError());
+        PGError error = p.error();
+        ASSERT_EQ(PGError::NO_SPACE_LEFT, error);
+    } else {
+        auto start_time = std::chrono::steady_clock::now();
+        bool res = true;
+        // follower need to wait for pg creation
+        while (!pg_exist(pg_id)) {
+            auto current_time = std::chrono::steady_clock::now();
+            auto duration = std::chrono::duration_cast<std::chrono::seconds>(current_time - start_time).count();
+            if (duration >= 20) {
+                LOGINFO("Failed to create pg {} at follower", pg_id);
+                res = false;
+                break;
+            }
+            std::this_thread::sleep_for(std::chrono::milliseconds(1000));
+        }
+        ASSERT_FALSE(res);
+        
+    }
+}
+
 TEST_F(HomeObjectFixture, PGRecoveryTest) {
     // create 10 pg
     for (pg_id_t i = 1; i < 11; i++) {
diff --git a/src/lib/homestore_backend/tests/test_heap_chunk_selector.cpp b/src/lib/homestore_backend/tests/test_heap_chunk_selector.cpp
index 0f3a1c1f..fb867664 100644
--- a/src/lib/homestore_backend/tests/test_heap_chunk_selector.cpp
+++ b/src/lib/homestore_backend/tests/test_heap_chunk_selector.cpp
@@ -40,7 +40,6 @@ class Chunk : public std::enable_shared_from_this< Chunk > {
 
     blk_num_t get_total_blks() const { return m_available_blks; }
     void set_chunk_id(uint16_t chunk_id) { m_chunk_id = chunk_id; }
-    const std::shared_ptr< Chunk > get_internal_chunk() { return shared_from_this(); }
     uint64_t size() const { return 1 * Mi; }
 
     Chunk(uint32_t pdev_id, uint16_t chunk_id, uint32_t available_blks, uint32_t defrag_nblks) {
@@ -75,15 +74,19 @@ blk_num_t VChunk::get_total_blks() const { return m_internal_chunk->get_total_bl
 
 uint64_t VChunk::size() const { return m_internal_chunk->size(); }
 
-cshared< Chunk > VChunk::get_internal_chunk() const { return m_internal_chunk->get_internal_chunk(); }
+cshared< Chunk > VChunk::get_internal_chunk() const { return m_internal_chunk; }
 
 } // namespace homestore
 
 using homeobject::csharedChunk;
 using homeobject::HeapChunkSelector;
+using homeobject::pg_id_t;
 using homestore::Chunk;
 using homestore::chunk_num_t;
 
+const pg_id_t FAKE_PG_ID = UINT16_MAX;
+const chunk_num_t FAKE_CHUNK_ID = UINT16_MAX;
+
 class HeapChunkSelectorTest : public ::testing::Test {
 protected:
     void SetUp() override {
@@ -112,23 +115,26 @@ class HeapChunkSelectorTest : public ::testing::Test {
             ASSERT_EQ(pg_heap_it->second->size(), 3);
 
             // test chunk_map
-            auto v2r_chunk_map_it = HCS.m_v2r_chunk_map.find(pg_id);
-            ASSERT_NE(v2r_chunk_map_it, HCS.m_v2r_chunk_map.end());
-            ASSERT_EQ(v2r_chunk_map_it->second->size(), 3);
+            auto v2p_chunk_map_it = HCS.m_v2p_chunk_map.find(pg_id);
+            ASSERT_NE(v2p_chunk_map_it, HCS.m_v2p_chunk_map.end());
+            ASSERT_EQ(v2p_chunk_map_it->second->size(), 3);
+
+            auto p2v_chunk_map_it = HCS.m_p2v_chunk_map.find(pg_id);
+            ASSERT_NE(p2v_chunk_map_it, HCS.m_p2v_chunk_map.end());
+            ASSERT_EQ(p2v_chunk_map_it->second->size(), 3);
 
-            auto r2v_chunk_map_it = HCS.m_r2v_chunk_map.find(pg_id);
-            ASSERT_NE(r2v_chunk_map_it, HCS.m_r2v_chunk_map.end());
-            ASSERT_EQ(r2v_chunk_map_it->second->size(), 3);
             for (int i = 0; i < 3; ++i) {
-                auto r_chunk_id = v2r_chunk_map_it->second->at(i);
-                ASSERT_EQ(i, r2v_chunk_map_it->second->at(r_chunk_id));
-                auto pdev_id = HCS.m_chunks[r_chunk_id]->get_pdev_id();
+                // test p_chunk_id <-> v_chunk_id
+                auto p_chunk_id = v2p_chunk_map_it->second->at(i);
+                ASSERT_EQ(i, p2v_chunk_map_it->second->at(p_chunk_id));
+                // test pg chunks must belong to same pdev
+                auto pdev_id = HCS.m_chunks[p_chunk_id]->get_pdev_id();
                 if (last_pdev_id != 0) {
                     ASSERT_EQ(last_pdev_id, pdev_id);
                 } else {
                     last_pdev_id = pdev_id;
                 }
-
+                // pdev heap should be empty at this point because all chunks have already been given to pg.
                 auto pdev_it = HCS.m_per_dev_heap.find(pdev_id);
                 ASSERT_NE(pdev_it, HCS.m_per_dev_heap.end());
                 ASSERT_EQ(pdev_it->second->size(), 0);
@@ -136,7 +142,6 @@ class HeapChunkSelectorTest : public ::testing::Test {
         }
     }
 
-
 public:
     HeapChunkSelector HCS;
 };
@@ -147,6 +152,35 @@ TEST_F(HeapChunkSelectorTest, test_for_each_chunk) {
     ASSERT_EQ(size.load(), 18);
 }
 
+TEST_F(HeapChunkSelectorTest, test_identical_layout) {
+    const homestore::blk_count_t count = 1;
+    homestore::blk_alloc_hints hints;
+    for (uint16_t pg_id = 1; pg_id < 4; ++pg_id) {
+        chunk_num_t p_chunk_id = 0;
+        for (int j = 3; j > 0; --j) {
+            const auto p_chunkID = HCS.peek_top_chunk(pg_id);
+            ASSERT_TRUE(p_chunkID.has_value());
+            const auto v_chunkID = HCS.get_virtual_chunk_id(pg_id, p_chunkID.value());
+            ASSERT_TRUE(v_chunkID.has_value());
+            auto chunk_id = HCS.get_physical_chunk_id(pg_id, v_chunkID.value());
+            ASSERT_EQ(p_chunkID.value(), chunk_id.value());
+            p_chunk_id = chunk_id.value();
+            hints.pdev_id_hint = ((uint32_t)pg_id << 16) | p_chunk_id;
+
+            ASSERT_NE(HCS.select_chunk(count, hints), nullptr);
+        }
+
+        // error handling test
+        ASSERT_FALSE(HCS.get_virtual_chunk_id(FAKE_PG_ID, 0).has_value());
+        ASSERT_FALSE(HCS.get_physical_chunk_id(FAKE_PG_ID, p_chunk_id).has_value());
+
+        ASSERT_FALSE(HCS.get_virtual_chunk_id(pg_id, FAKE_CHUNK_ID).has_value());
+        ASSERT_FALSE(HCS.get_physical_chunk_id(pg_id, FAKE_CHUNK_ID).has_value());
+        // all chunks have been given out
+        ASSERT_FALSE(HCS.peek_top_chunk(pg_id).has_value());
+    }
+}
+
 TEST_F(HeapChunkSelectorTest, test_select_chunk) {
     homestore::blk_count_t count = 1;
     homestore::blk_alloc_hints hints;
@@ -154,84 +188,80 @@ TEST_F(HeapChunkSelectorTest, test_select_chunk) {
     ASSERT_EQ(chunk, nullptr);
 
     for (uint16_t pg_id = 1; pg_id < 4; ++pg_id) {
-        hints.pdev_id_hint = pg_id; // tmp bypass using pdev_id_hint present pg_id
         for (int j = 3; j > 0; --j) {
+            chunk_num_t p_chunk_id = j + 3 * (pg_id - 1);
+            hints.pdev_id_hint = ((uint32_t)pg_id << 16) | p_chunk_id;
             auto chunk = HCS.select_chunk(count, hints);
             ASSERT_NE(chunk, nullptr);
-            ASSERT_EQ(chunk->get_pdev_id(), pg_id);
+            ASSERT_EQ(chunk->get_pdev_id(), pg_id); // in this ut, pg_id is same as pdev id
             ASSERT_EQ(chunk->available_blks(), j);
         }
     }
 }
 
-
 TEST_F(HeapChunkSelectorTest, test_select_specific_chunk) {
-    const uint16_t pg_id = 1;
-    auto chunk_ids = HCS.get_pg_chunks(pg_id);
-    ASSERT_NE(chunk_ids, nullptr);
-    const chunk_num_t chunk_id = chunk_ids->at(0);
-
-    auto chunk = HCS.select_specific_chunk(pg_id, chunk_id);
-    ASSERT_EQ(chunk->get_chunk_id(), chunk_id);
-    auto pdev_id = chunk->get_pdev_id();
-
-    // make sure pg chunk map
-    auto pg_heap_it = HCS.m_per_pg_heap.find(pg_id);
-    ASSERT_NE(pg_heap_it, HCS.m_per_pg_heap.end());
-    ASSERT_EQ(pg_heap_it->second->size(), 2);
-
-    // test chunk_map stable
-    auto v2r_chunk_map_it = HCS.m_v2r_chunk_map.find(pg_id);
-    ASSERT_NE(v2r_chunk_map_it, HCS.m_v2r_chunk_map.end());
-    ASSERT_EQ(v2r_chunk_map_it->second->size(), 3);
-
-    auto r2v_chunk_map_it = HCS.m_r2v_chunk_map.find(pg_id);
-    ASSERT_NE(r2v_chunk_map_it, HCS.m_r2v_chunk_map.end());
-    ASSERT_EQ(r2v_chunk_map_it->second->size(), 3);
-
-    // select the rest chunks to make sure specific chunk does not exist in HeapChunkSelector anymore.
-    homestore::blk_count_t count = 1;
-    homestore::blk_alloc_hints hints;
-    hints.pdev_id_hint = pg_id;
-    for (int j = 2; j > 0; --j) {
-        auto chunk = HCS.select_chunk(count, hints);
-        ASSERT_EQ(chunk->get_pdev_id(), pdev_id);
+    for (uint16_t pg_id = 1; pg_id < 4; ++pg_id) {
+        auto chunk_ids = HCS.get_pg_chunks(pg_id);
+        ASSERT_NE(chunk_ids, nullptr);
+        const chunk_num_t chunk_id = chunk_ids->at(0);
+
+        auto chunk = HCS.select_specific_chunk(pg_id, chunk_id);
+        ASSERT_EQ(chunk->get_chunk_id(), chunk_id);
+
+        auto pg_heap_it = HCS.m_per_pg_heap.find(pg_id);
+        ASSERT_NE(pg_heap_it, HCS.m_per_pg_heap.end());
+        ASSERT_EQ(pg_heap_it->second->size(), 2);
+
+        // test chunk_map stable
+        auto v2p_chunk_map_it = HCS.m_v2p_chunk_map.find(pg_id);
+        ASSERT_NE(v2p_chunk_map_it, HCS.m_v2p_chunk_map.end());
+        ASSERT_EQ(v2p_chunk_map_it->second->size(), 3);
+
+        auto p2v_chunk_map_it = HCS.m_p2v_chunk_map.find(pg_id);
+        ASSERT_NE(p2v_chunk_map_it, HCS.m_p2v_chunk_map.end());
+        ASSERT_EQ(p2v_chunk_map_it->second->size(), 3);
+
+        // release this chunk to HeapChunkSelector and try again
+        HCS.release_chunk(pg_id, chunk_id);
+        chunk = HCS.select_specific_chunk(pg_id, chunk_id);
+        ASSERT_EQ(pg_id, chunk->get_pdev_id());
+        ASSERT_EQ(chunk_id, chunk->get_chunk_id());
     }
-
-    // release this chunk to HeapChunkSelector
-    HCS.release_chunk(pg_id, chunk_id);
-    chunk = HCS.select_chunk(1, hints);
-    ASSERT_EQ(1, chunk->get_pdev_id());
-    ASSERT_EQ(chunk_id, chunk->get_chunk_id());
-
 }
 
-
 TEST_F(HeapChunkSelectorTest, test_release_chunk) {
-    homestore::blk_count_t count = 1;
-    homestore::blk_alloc_hints hints;
-    const uint16_t pg_id = 1;
-    hints.pdev_id_hint = pg_id;
-    auto chunk1 = HCS.select_chunk(count, hints);
-    auto pdev_id = chunk1->get_pdev_id();
-
-    ASSERT_EQ(chunk1->get_pdev_id(), pdev_id);
-    ASSERT_EQ(chunk1->available_blks(), 3);
-
-    auto chunk2 = HCS.select_chunk(count, hints);
-    ASSERT_EQ(chunk2->get_pdev_id(), pdev_id);
-    ASSERT_EQ(chunk2->available_blks(), 2);
-
-    HCS.release_chunk(pg_id, chunk1->get_chunk_id());
-    HCS.release_chunk(pg_id, chunk2->get_chunk_id());
-
-    chunk1 = HCS.select_chunk(count, hints);
-    ASSERT_EQ(chunk1->get_pdev_id(), pdev_id);
-    ASSERT_EQ(chunk1->available_blks(), 3);
-
-    chunk2 = HCS.select_chunk(count, hints);
-    ASSERT_EQ(chunk2->get_pdev_id(), pdev_id);
-    ASSERT_EQ(chunk2->available_blks(), 2);
+    for (uint16_t pg_id = 1; pg_id < 4; ++pg_id) {
+        auto p_chunkID = HCS.peek_top_chunk(pg_id);
+        ASSERT_TRUE(p_chunkID.has_value());
+        const auto p_chunk_id_1 = p_chunkID.value();
+        auto chunk1 = HCS.select_specific_chunk(pg_id, p_chunk_id_1);
+        ASSERT_NE(chunk1, nullptr);
+        auto pdev_id = chunk1->get_pdev_id();
+        ASSERT_EQ(chunk1->get_pdev_id(), pdev_id);
+        ASSERT_EQ(chunk1->available_blks(), 3);
+
+        p_chunkID = HCS.peek_top_chunk(pg_id);
+        ASSERT_TRUE(p_chunkID.has_value());
+        const auto p_chunk_id_2 = p_chunkID.value();
+        auto chunk2 = HCS.select_specific_chunk(pg_id, p_chunk_id_2);
+        ASSERT_NE(chunk2, nullptr);
+        ASSERT_EQ(chunk2->get_pdev_id(), pdev_id);
+        ASSERT_EQ(chunk2->available_blks(), 2);
+
+        ASSERT_TRUE(HCS.release_chunk(pg_id, chunk1->get_chunk_id()));
+        ASSERT_TRUE(HCS.release_chunk(pg_id, chunk2->get_chunk_id()));
+
+        chunk1 = HCS.select_specific_chunk(pg_id, p_chunk_id_1);
+        ASSERT_EQ(chunk1->get_pdev_id(), pdev_id);
+        ASSERT_EQ(chunk1->available_blks(), 3);
+
+        chunk2 = HCS.select_specific_chunk(pg_id, p_chunk_id_2);
+        ASSERT_EQ(chunk2->get_pdev_id(), pdev_id);
+        ASSERT_EQ(chunk2->available_blks(), 2);
+
+        ASSERT_FALSE(HCS.release_chunk(FAKE_PG_ID, FAKE_CHUNK_ID));
+        ASSERT_FALSE(HCS.release_chunk(pg_id, FAKE_CHUNK_ID));
+    }
 }
 
 TEST_F(HeapChunkSelectorTest, test_recovery) {
@@ -242,49 +272,75 @@ TEST_F(HeapChunkSelectorTest, test_recovery) {
     HCS_recovery.add_chunk(std::make_shared< Chunk >(2, 4, 1, 6));
     HCS_recovery.add_chunk(std::make_shared< Chunk >(2, 5, 2, 5));
     HCS_recovery.add_chunk(std::make_shared< Chunk >(2, 6, 3, 4));
+    HCS_recovery.add_chunk(std::make_shared< Chunk >(3, 7, 1, 3));
+    HCS_recovery.add_chunk(std::make_shared< Chunk >(3, 8, 2, 2));
+    HCS_recovery.add_chunk(std::make_shared< Chunk >(3, 9, 3, 1));
+
+    // on_pg_meta_blk_found    
+    for (uint16_t pg_id = 1; pg_id < 4; ++pg_id) {
+        std::vector< chunk_num_t > chunk_ids{1, 2};
+        std::vector< chunk_num_t > chunk_ids_for_twice{1, 2};
+        std::vector< chunk_num_t > chunk_ids_not_valid{1, 20};
+        std::vector< chunk_num_t > chunk_ids_not_same_pdev{1, 6};
+        for (chunk_num_t j = 0; j < 2; ++j) {
+            chunk_ids[j] += (pg_id - 1) * 3;
+            chunk_ids_for_twice[j] += (pg_id - 1) * 3;
+            chunk_ids_not_valid[j] += (pg_id - 1) * 3;
+            chunk_ids_not_same_pdev[j] += ((pg_id - 1) * 3) % 9;
+        }
 
-    std::vector<chunk_num_t> chunk_ids {1,2,3};
-    const uint16_t pg_id = 1;
-    // test recover chunk map
-    HCS_recovery.set_pg_chunks(pg_id, std::move(chunk_ids));
-    auto v2r_chunk_map_it = HCS_recovery.m_v2r_chunk_map.find(pg_id);
-    ASSERT_NE(v2r_chunk_map_it, HCS_recovery.m_v2r_chunk_map.end());
-    ASSERT_EQ(v2r_chunk_map_it->second->size(), 3);
-
-    auto r2v_chunk_map_it = HCS_recovery.m_r2v_chunk_map.find(pg_id);
-    ASSERT_NE(r2v_chunk_map_it, HCS_recovery.m_r2v_chunk_map.end());
-    ASSERT_EQ(r2v_chunk_map_it->second->size(), 3);
-    // test recover pdev map
+        // test recover chunk map
+        ASSERT_FALSE(HCS_recovery.set_pg_chunks(pg_id, std::move(chunk_ids_not_valid)));
+        ASSERT_FALSE(HCS_recovery.set_pg_chunks(pg_id, std::move(chunk_ids_not_same_pdev)));
+
+        ASSERT_TRUE(HCS_recovery.set_pg_chunks(pg_id, std::move(chunk_ids)));
+        // can't set pg chunks twice
+        ASSERT_FALSE(HCS_recovery.set_pg_chunks(pg_id, std::move(chunk_ids_for_twice)));
+
+        auto v2p_chunk_map_it = HCS_recovery.m_v2p_chunk_map.find(pg_id);
+        ASSERT_NE(v2p_chunk_map_it, HCS_recovery.m_v2p_chunk_map.end());
+        ASSERT_EQ(v2p_chunk_map_it->second->size(), 2);
+
+        auto p2v_chunk_map_it = HCS_recovery.m_p2v_chunk_map.find(pg_id);
+        ASSERT_NE(p2v_chunk_map_it, HCS_recovery.m_p2v_chunk_map.end());
+        ASSERT_EQ(p2v_chunk_map_it->second->size(), 2);
+    }
+
+    //  on_pg_meta_blk_recover_completed
     HCS_recovery.recover_per_dev_chunk_heap();
-    auto pdev_it = HCS_recovery.m_per_dev_heap.find(1);
-    ASSERT_NE(pdev_it, HCS_recovery.m_per_dev_heap.end());
-    ASSERT_EQ(pdev_it->second->size(), 0);
-
-    pdev_it = HCS_recovery.m_per_dev_heap.find(2);
-    ASSERT_NE(pdev_it, HCS_recovery.m_per_dev_heap.end());
-    ASSERT_EQ(pdev_it->second->size(), 3);
-    auto &pdev_heap = pdev_it->second->m_heap;
-    auto vchunk = homestore::VChunk(nullptr);
-    for (int i = 6; i > 3; --i) {
+    for (uint16_t pg_id = 1; pg_id < 4; ++pg_id) {
+        // test recover pdev map size
+        auto pdev_it = HCS_recovery.m_per_dev_heap.find(pg_id);
+        ASSERT_NE(pdev_it, HCS_recovery.m_per_dev_heap.end());
+        ASSERT_EQ(pdev_it->second->size(), 1); // 1 = 3(all) - 2(pg)
+
+        auto& pdev_heap = pdev_it->second->m_heap;
+        auto vchunk = homestore::VChunk(nullptr);
         vchunk = pdev_heap.top();
-        pdev_heap.pop();
-        ASSERT_EQ(vchunk.get_chunk_id(), i);
+        ASSERT_EQ(vchunk.get_chunk_id(), 3 + (pg_id - 1) * 3);
     }
 
-    // test recover pg heap
-    std::unordered_set< homestore::chunk_num_t > excluding_chunks;
-    excluding_chunks.emplace(1);
-    HCS_recovery.recover_pg_chunk_heap(pg_id, excluding_chunks);
-    auto pg_heap_it = HCS_recovery.m_per_pg_heap.find(pg_id);
-    ASSERT_NE(pg_heap_it, HCS_recovery.m_per_pg_heap.end());
-    ASSERT_EQ(pg_heap_it->second->size(), 2);
-
-    homestore::blk_alloc_hints hints;
-    hints.pdev_id_hint = pg_id;
-    for (int j = 3; j > 1; --j) {
-        auto chunk = HCS_recovery.select_chunk(1, hints);
-        ASSERT_EQ(chunk->get_pdev_id(), 1);
-        ASSERT_EQ(chunk->available_blks(), j);
+    // on_shard_meta_blk_recover_completed
+    for (uint16_t pg_id = 1; pg_id < 4; ++pg_id) {
+        // test recover pg heap
+        std::unordered_set< homestore::chunk_num_t > excluding_chunks;
+        excluding_chunks.emplace(1 + (pg_id - 1) * 3);
+
+        ASSERT_FALSE(HCS_recovery.recover_pg_chunk_heap(FAKE_PG_ID, excluding_chunks));
+        ASSERT_TRUE(HCS_recovery.recover_pg_chunk_heap(pg_id, excluding_chunks));
+        //can't recover pg chunk heap twice
+        ASSERT_FALSE(HCS_recovery.recover_pg_chunk_heap(pg_id, excluding_chunks));
+
+        auto pg_heap_it = HCS_recovery.m_per_pg_heap.find(pg_id);
+        ASSERT_NE(pg_heap_it, HCS_recovery.m_per_pg_heap.end());
+        ASSERT_EQ(pg_heap_it->second->size(), 1);
+
+        const auto p_chunkID = HCS_recovery.peek_top_chunk(pg_id);
+        ASSERT_TRUE(p_chunkID.has_value());
+        auto chunk = HCS_recovery.select_specific_chunk(pg_id, p_chunkID.value());
+        ASSERT_NE(chunk, nullptr);
+        ASSERT_EQ(chunk->get_pdev_id(), pg_id);
+        ASSERT_EQ(chunk->available_blks(), 2);  
     }
 }