diff --git a/conanfile.py b/conanfile.py index b94e809a..375bb53f 100644 --- a/conanfile.py +++ b/conanfile.py @@ -10,7 +10,7 @@ class HomeObjectConan(ConanFile): name = "homeobject" - version = "4.3.0" + version = "4.3.1" homepage = "https://github.com/eBay/HomeObject" description = "Blob Store built on HomeStore" diff --git a/src/lib/homestore_backend/hs_shard_manager.cpp b/src/lib/homestore_backend/hs_shard_manager.cpp index cd243039..99b88682 100644 --- a/src/lib/homestore_backend/hs_shard_manager.cpp +++ b/src/lib/homestore_backend/hs_shard_manager.cpp @@ -397,42 +397,10 @@ bool HSHomeObject::on_shard_message_pre_commit(int64_t lsn, sisl::blob const& he return false; } -#ifdef _PRERELEASE - if (msg_header->msg_type == ReplicationMessageType::SEAL_SHARD_MSG) { - // Pause SEAL pre_commit at function entry, before state=SEALED. Test thread can race - // _put_blob while shard is still OPEN; sealed_lsn guard rejects the late blob on commit. - iomgr_flip::instance()->callback_flip("pause_seal_pre_commit"); - } -#endif - - switch (msg_header->msg_type) { - case ReplicationMessageType::SEAL_SHARD_MSG: { - auto decoded_shard_sb = - decode_shard_sb(r_cast< const shard_info_superblk* >(header.cbytes() + sizeof(ReplicationMessageHeader))); - RELEASE_ASSERT(decoded_shard_sb.has_value(), "failed to decode shard superblk in pre_commit SEAL_SHARD_MSG"); - auto& [shard_info, p_chunk_id_unused, v_chunk_id_unused] = decoded_shard_sb.value(); - - { - std::scoped_lock lock_guard(_shard_lock); - auto iter = _shard_map.find(shard_info.id); - RELEASE_ASSERT(iter != _shard_map.end(), "shardID=0x{:x}, pg={}, shard=0x{:x}, shard does not exist", - shard_info.id, (shard_info.id >> homeobject::shard_width), - (shard_info.id & homeobject::shard_mask)); - auto& state = (*iter->second)->info.state; - // we just change the state to SEALED, so that it will fail the later coming put_blob on this shard and will - // be easy for rollback. - // the update of superblk will be done in on_shard_message_commit; - if (state == ShardInfo::State::OPEN) { - state = ShardInfo::State::SEALED; - } else { - SLOGW(tid, shard_info.id, "try to seal an unopened shard"); - } - } - } - default: { - break; - } - } + // SEAL_SHARD state transition is done in on_shard_message_commit, not here. + // Reason: create+seal can land in the same raft batch; create only inserts the shard into + // _shard_map on commit, so seal pre_commit would fail to find the shard. + // Late put_blob is rejected by the sealed_lsn guard in on_blob_put_commit. return true; } @@ -459,27 +427,9 @@ void HSHomeObject::on_shard_message_rollback(int64_t lsn, sisl::blob const& head break; } case ReplicationMessageType::SEAL_SHARD_MSG: { - auto decoded_shard_sb = - decode_shard_sb(r_cast< const shard_info_superblk* >(header.cbytes() + sizeof(ReplicationMessageHeader))); - RELEASE_ASSERT(decoded_shard_sb.has_value(), "failed to decode shard superblk in rollback SEAL_SHARD_MSG"); - auto& [shard_info, p_chunk_id_unused, v_chunk_id_unused] = decoded_shard_sb.value(); - { - std::scoped_lock lock_guard(_shard_lock); - auto iter = _shard_map.find(shard_info.id); - RELEASE_ASSERT(iter != _shard_map.end(), "shardID=0x{:x}, pg={}, shard=0x{:x}, shard does not exist", - shard_info.id, (shard_info.id >> homeobject::shard_width), - (shard_info.id & homeobject::shard_mask)); - auto& state = (*iter->second)->info.state; - // we just change the state to SEALED, since it will be easy for rollback - // the update of superblk will be done in on_shard_message_commit; - if (state == ShardInfo::State::SEALED) { - state = ShardInfo::State::OPEN; - SLOGD(tid, shard_info.id, "rollback seal shard message, lsn={}", lsn); - } else { - SLOGW(tid, shard_info.id, - "try to rollback seal_shard message , but the shard state is not sealed, lsn= {}", lsn); - } - } + // Seal no longer mutates shard state in pre_commit, so rollback has nothing to revert + // in the shard map. Just fail the proposer's promise. + SLOGD(tid, msg_header->shard_id, "rollback seal shard message, type={}, lsn={}", msg_header->msg_type, lsn); // TODO:set a proper error code if (ctx) { ctx->promise_.setValue(folly::makeUnexpected(ShardError(ShardErrorCode::RETRY_REQUEST))); } @@ -532,6 +482,10 @@ void HSHomeObject::on_shard_message_commit(int64_t lsn, sisl::blob const& h, sha #ifdef _PRERELEASE if (header->msg_type == ReplicationMessageType::SEAL_SHARD_MSG) { + // Pause SEAL commit at function entry, before state=SEALED / sealed_lsn update. + // Test thread can race _put_blob while shard is still OPEN; sealed_lsn guard rejects + // the late blob when this gate is released and seal commit finishes. + iomgr_flip::instance()->callback_flip("pause_seal_commit"); // Wait for CREATE_SHARD (next log) to be in log store before SEAL releases its vchunk. // Polled via get_last_append_lsn() so it doesn't rely on pre_commit signals. Armed on the // repro follower only; no-op on leader and other followers. @@ -567,19 +521,22 @@ void HSHomeObject::on_shard_message_commit(int64_t lsn, sisl::blob const& h, sha RELEASE_ASSERT(decoded_shard_sb.has_value(), "failed to decode shard superblk in commit SEAL_SHARD_MSG"); auto& [shard_info, p_chunk_id_unused, v_chunk_id_unused] = decoded_shard_sb.value(); - ShardInfo::State state; { std::scoped_lock lock_guard(_shard_lock); auto iter = _shard_map.find(shard_info.id); RELEASE_ASSERT(iter != _shard_map.end(), "shardID=0x{:x}, pg={}, shard=0x{:x}, shard does not exist", shard_info.id, (shard_info.id >> homeobject::shard_width), (shard_info.id & homeobject::shard_mask)); - state = (*iter->second)->info.state; + auto& state = (*iter->second)->info.state; + // State transition moved here from pre_commit so create+seal in the same raft batch works: + // create commit adds the shard to the map before seal commit runs. + if (state == ShardInfo::State::OPEN) { + state = ShardInfo::State::SEALED; + } else { + SLOGW(tid, shard_info.id, "try to seal an unopened shard, current_state={}", static_cast< int >(state)); + } } - RELEASE_ASSERT(state == ShardInfo::State::SEALED, - "try to commit SEAL_SHARD_MSG but shard state is not sealed. shardID={}", shard_info.id); - // Corner case: // Assume cp_lsn = dc_lsn = 10. // lsn 11: put_blob (blob -> pba in chunk-1) diff --git a/src/lib/homestore_backend/tests/hs_gc_tests.cpp b/src/lib/homestore_backend/tests/hs_gc_tests.cpp index 450ff101..72d3c36a 100644 --- a/src/lib/homestore_backend/tests/hs_gc_tests.cpp +++ b/src/lib/homestore_backend/tests/hs_gc_tests.cpp @@ -1094,25 +1094,25 @@ TEST_F(HomeObjectFixture, StalePChunkRouteAfterGC) { } // =================================================================================================== -// StaleBlobRouteAfterSealAndGC: PUT_BLOB races with SEAL_SHARD pre_commit; sealed_lsn guard rejects it. +// StaleBlobRouteAfterSealAndGC: PUT_BLOB races with SEAL_SHARD commit; sealed_lsn guard rejects it. // // Production scenario : // A PUT_BLOB whose admission check passed (shard OPEN) should be rejected if the shard gets // sealed before the blob is committed. The sealed_lsn guard in on_blob_put_commit must catch it. // // Exact sequence modelled (single replica, leader): -// ① SEAL_SHARD pre_commit fires and PAUSES before changing state=SEALED -// (flip "pause_seal_pre_commit"). At this point shard state is still OPEN. +// ① SEAL_SHARD commit fires and PAUSES before changing state=SEALED / sealed_lsn +// (flip "pause_seal_commit"). At this point shard state is still OPEN. // ② _put_blob is called in the test thread. get_blk_alloc_hints sees state==OPEN → passes, // blk is allocated on pchunk_A. The put is async (raft not yet committed). -// ③ Gate releases → state = SEALED → SEAL_SHARD commit → sealed_lsn = X. +// ③ Gate releases → state = SEALED → sealed_lsn = X. // ④ PUT_BLOB commit (lsn = X+1): on_blob_put_commit checks lsn(X+1) >= sealed_lsn(X) → reject. // The allocated blk is freed; the blob does NOT land in the pg index. // // Verification: the late blob is absent from the index; bulk blobs are still readable. // // Runs on the leader replica only; no multi-replica complexity needed. -// Pause point: flip "pause_seal_pre_commit". Compiled out of release builds. +// Pause point: flip "pause_seal_commit". Compiled out of release builds. // =================================================================================================== TEST_F(HomeObjectFixture, StaleBlobRouteAfterSealAndGC) { const pg_id_t pg_id = 1; @@ -1141,23 +1141,22 @@ TEST_F(HomeObjectFixture, StaleBlobRouteAfterSealAndGC) { std::map< pg_id_t, blob_id_t > pg_blob_id{{pg_id, 0}}; put_blobs(shards, num_blobs_per_shard, pg_blob_id); - // ---- arm the flip on the leader: pause SEAL pre_commit before state=SEALED ---- + // ---- arm the flip on the leader: pause SEAL commit before state=SEALED ---- if (i_am_leader) { auto dont_care = m_fc.create_condition("", flip::Operator::DONT_CARE, (int)0); flip::FlipFrequency freq; freq.set_count(3); // 3 replicas all call callback_flip; count must be >= num_replicas freq.set_percent(100); - m_fc.inject_callback_flip< void >("pause_seal_pre_commit", {dont_care}, freq, std::function< void() >([&]() { - LOGI( - "[StaleBlobRouteAfterSealAndGC] pausing SEAL pre_commit BEFORE lock"); - std::unique_lock< std::mutex > lk(repro2_mtx); - repro2_blocked.store(true); - repro2_cv.notify_all(); - repro2_cv.wait(lk, [&] { return repro2_released.load(); }); - LOGI("[StaleBlobRouteAfterSealAndGC] resuming SEAL pre_commit"); - })); - LOGINFO("[StaleBlobRouteAfterSealAndGC] armed pause_seal_pre_commit on leader replica={}", - g_helper->replica_num()); + m_fc.inject_callback_flip< void >( + "pause_seal_commit", {dont_care}, freq, std::function< void() >([&]() { + LOGI("[StaleBlobRouteAfterSealAndGC] pausing SEAL commit BEFORE state update"); + std::unique_lock< std::mutex > lk(repro2_mtx); + repro2_blocked.store(true); + repro2_cv.notify_all(); + repro2_cv.wait(lk, [&] { return repro2_released.load(); }); + LOGI("[StaleBlobRouteAfterSealAndGC] resuming SEAL commit"); + })); + LOGINFO("[StaleBlobRouteAfterSealAndGC] armed pause_seal_commit on leader replica={}", g_helper->replica_num()); } g_helper->sync(); // make sure flip is armed on all replicas before proceeding @@ -1166,7 +1165,7 @@ TEST_F(HomeObjectFixture, StaleBlobRouteAfterSealAndGC) { blob_id_t late_blob_id [[maybe_unused]] = INVALID_UINT64_ID; if (i_am_leader) { // 1. Start seal_shard in a background thread so it runs concurrently. - // seal_shard will hit the gate in pre_commit and pause there. + // seal_shard will hit the gate in commit and pause there. auto tid = generateRandomTraceId(); bool seal_ok = false; std::thread seal_thread([&]() { @@ -1174,18 +1173,18 @@ TEST_F(HomeObjectFixture, StaleBlobRouteAfterSealAndGC) { seal_ok = r.hasValue(); }); - // 2. Wait until pre_commit is paused (shard state is still OPEN). + // 2. Wait until commit is paused (shard state is still OPEN). { std::unique_lock< std::mutex > lk(repro2_mtx); if (!repro2_cv.wait_for(lk, std::chrono::seconds(30), [&] { return repro2_blocked.load(); })) { repro2_released.store(true); // avoid deadlock if gate never fires repro2_cv.notify_all(); seal_thread.join(); - m_fc.remove_flip("pause_seal_pre_commit"); - FAIL() << "SEAL pre_commit never reached the pause point"; + m_fc.remove_flip("pause_seal_commit"); + FAIL() << "SEAL commit never reached the pause point"; } } - LOGINFO("[StaleBlobRouteAfterSealAndGC] leader sees SEAL pre_commit paused; shard state=OPEN; " + LOGINFO("[StaleBlobRouteAfterSealAndGC] leader sees SEAL commit paused; shard state=OPEN; " "calling _put_blob with shard still OPEN"); // 3. Call _put_blob in a background thread: shard state is OPEN → get_blk_alloc_hints @@ -1199,9 +1198,8 @@ TEST_F(HomeObjectFixture, StaleBlobRouteAfterSealAndGC) { b.hasValue() ? "admitted" : "rejected"); }); - // 4. Release the gate: state = SEALED, seal pre_commit returns → raft commits seal. - // After seal commit, sealed_lsn = lsn_seal. Then put_blob commit fires and - // on_blob_put_commit checks lsn(put) >= sealed_lsn → rejects. + // 4. Release the gate: state = SEALED, seal commit returns with sealed_lsn = lsn_seal. + // Then put_blob commit fires and on_blob_put_commit checks lsn(put) >= sealed_lsn → rejects. { std::unique_lock< std::mutex > lk(repro2_mtx); repro2_released.store(true); @@ -1212,7 +1210,7 @@ TEST_F(HomeObjectFixture, StaleBlobRouteAfterSealAndGC) { // 5. Wait for both background threads. blob_thread.join(); seal_thread.join(); - m_fc.remove_flip("pause_seal_pre_commit"); + m_fc.remove_flip("pause_seal_commit"); ASSERT_TRUE(seal_ok) << "seal_shard failed"; EXPECT_TRUE(blob_rejected)