update

src/lib/homestore_backend/hs_homeobject.hpp

@@ -161,7 +161,8 @@ class HSHomeObject : public HomeObjectImpl {
     shared< HeapChunkSelector > chunk_selector_;
     iomgr::timer_handle_t ho_timer_thread_handle_;
 
-    // used for testing
+    // used for testing when we only have solo_repl_dev.
+    // TODO: remove or change this when we have raft_repl_dev.
     shared< std::binary_semaphore > smphSignal;
 
 private:
@@ -243,6 +244,8 @@ class HSHomeObject : public HomeObjectImpl {
 
     void trigger_timed_events();
 
+    // used for testing when we only have solo_repl_dev.
+    // TODO: remove or change this when we have raft_repl_dev.
     void set_semaphore();
     void release_semaphore();
 };

src/lib/homestore_backend/hs_shard_manager.cpp

@@ -179,8 +179,9 @@ bool HSHomeObject::do_shard_message_pre_commit(int64_t lsn, ReplicationMessageHe
     auto shard_info = deserialize_shard_info(r_cast< const char* >(value.bytes), value.size);
     switch (header.msg_type) {
     case ReplicationMessageType::SEAL_SHARD_MSG: {
-        // in pre commit, we just update shard info in map and return to client w/o writing metablk;
-        // since it will be easy to rollback if failing to reach consensus .
+        // we can not release chunk here, since if rollback happens, we can not make sure we can get the same chunk.
+        // it might be selected by other creat_shard after we release it.
+        // we need to wait for the commit phase to release chunk;
         update_shard_in_map(shard_info);
         if (ctx) { ctx->promise_.setValue(ShardManager::Result< ShardInfo >(shard_info)); }
         break;
@@ -314,10 +315,22 @@ void HSHomeObject::do_shard_message_commit(int64_t lsn, ReplicationMessageHeader
         break;
     }
     case ReplicationMessageType::SEAL_SHARD_MSG: {
+        ShardInfo::State state;
+        {
+            std::scoped_lock lock_guard(_shard_lock);
+            auto iter = _shard_map.find(shard_info.id);
+            RELEASE_ASSERT(iter != _shard_map.end(), "Missing shard info");
+            state = (*iter->second)->info.state;
+        }
+
+        RELEASE_ASSERT(state == ShardInfo::State::SEALED, "Shard should be sealed before commit");
+
         auto chunk_id = get_shard_chunk(shard_info.id);
         RELEASE_ASSERT(chunk_id.has_value(), "Chunk id not found");
+        // when restarting, the chunk is not selected since the metablk indicates the shard is sealed.
+        // fix me!!!!
+        chunk_selector()->select_specific_chunk(chunk_id.value());
         chunk_selector()->release_chunk(chunk_id.value());
-        break;
     }
     default: {
         break;