From 1cd6bfcdd80c8e62c82c4bf1e31000df02161e0d Mon Sep 17 00:00:00 2001 From: "Kou, Jilong" Date: Tue, 14 Jul 2026 11:46:32 +0800 Subject: [PATCH 1/2] Fix missed pg_destroy input in #435 --- src/lib/homestore_backend/replication_state_machine.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/lib/homestore_backend/replication_state_machine.cpp b/src/lib/homestore_backend/replication_state_machine.cpp index 3ae6cfee..d2100fb9 100644 --- a/src/lib/homestore_backend/replication_state_machine.cpp +++ b/src/lib/homestore_backend/replication_state_machine.cpp @@ -469,7 +469,7 @@ void ReplicationStateMachine::write_snapshot_obj(std::shared_ptr< homestore::sna if (home_object_->pg_exists(pg_data->pg_id())) { LOGI("pg already exists, clean pg resources before snapshot, pg={} {}", pg_data->pg_id(), log_suffix); // Need to pause state machine before destroying the PG, if fail, let raft retry. - if (!home_object_->pg_destroy(pg_data->pg_id())) { + if (!home_object_->pg_destroy(pg_data->pg_id(), true)) { LOGE("failed to destroy existing pg, let raft retry, pg={} {}", pg_data->pg_id(), log_suffix); return; } From 3066fedca935d4ca8b13b0805efdf9adebc58511 Mon Sep 17 00:00:00 2001 From: "Kou, Jilong" Date: Tue, 14 Jul 2026 11:50:42 +0800 Subject: [PATCH 2/2] Fix data corruption on CP persistence inconsistency During BR the IndexSvc journal and the AppendBlkAllocator CP superblock are flushed in parallel with no atomicity guarantee. If the process crashes after the index entry is durable but before the allocator superblock is written, the receiver resumes with the blk_id visible in the index but the allocator watermarks reset. The dedup skip path then bypasses commit_blk entirely, leaving the blocks silently available for reallocation by subsequent blob writes and corrupting the data on disk. The fix re-commits the blk_id recovered from the index entry so the allocator watermarks are restored before new allocations proceed. --- .gitignore | 1 + conanfile.py | 4 +- .../snapshot_receive_handler.cpp | 14 +- .../tests/homeobj_misc_tests.cpp | 209 +++++++++++++++++- 4 files changed, 221 insertions(+), 7 deletions(-) diff --git a/.gitignore b/.gitignore index c02f7996..2f49dfde 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,7 @@ .cache .clangd .claude/* +.opencode/* build test_package/build diff --git a/conanfile.py b/conanfile.py index c6f83277..1840dbe7 100644 --- a/conanfile.py +++ b/conanfile.py @@ -10,7 +10,7 @@ class HomeObjectConan(ConanFile): name = "homeobject" - version = "4.2.5" + version = "4.2.6" homepage = "https://github.com/eBay/HomeObject" description = "Blob Store built on HomeStore" @@ -51,7 +51,7 @@ def build_requirements(self): def requirements(self): self.requires("sisl/[^13.2]", transitive_headers=True) - self.requires("homestore/[^7.5.10]") + self.requires("homestore/[^7.5.12]") def validate(self): if self.info.settings.compiler.cppstd: diff --git a/src/lib/homestore_backend/snapshot_receive_handler.cpp b/src/lib/homestore_backend/snapshot_receive_handler.cpp index bd3ab3be..eacfb264 100644 --- a/src/lib/homestore_backend/snapshot_receive_handler.cpp +++ b/src/lib/homestore_backend/snapshot_receive_handler.cpp @@ -151,7 +151,6 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob #ifdef _PRERELEASE auto delay = iomgr_flip::instance()->get_test_flip< long >("simulate_write_snapshot_save_blob_delay", static_cast< long >(blob->blob_id())); - LOGD("simulate_write_snapshot_save_blob_delay flip, triggered={}, blob={}", delay.has_value(), blob->blob_id()); if (delay) { LOGI("Simulating pg snapshot receive data with delay, delay={}, blob_id={}", delay.get(), blob->blob_id()); std::this_thread::sleep_for(std::chrono::milliseconds(delay.get())); @@ -165,8 +164,17 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob ctx_->index_table = hs_pg->index_table_; } RELEASE_ASSERT(ctx_->index_table != nullptr, "Index table instance null"); - if (home_obj_.get_blob_from_index_table(ctx_->index_table, ctx_->shard_cursor, blob->blob_id())) { - LOGD("Skip already persisted blob_id={}", blob->blob_id()); + if (auto blob_index = + home_obj_.get_blob_from_index_table(ctx_->index_table, ctx_->shard_cursor, blob->blob_id()); + blob_index) { + // Blob already in index, re-commit the blk_id to restore both allocator watermarks (m_last_append_offset + // and m_commit_offset) in case we are recovering from a crash where the IndexSvc txn journal persisted the + // BTree entry but the AppendBlkAllocator CP superblock was not yet flushed. Without this the allocator + // would consider those blocks free and could silently overwrite them with a new blob, causing data + // corruption. + auto blk_id = blob_index.value(); + homestore::data_service().commit_blk(blk_id, true /* recommit */); + LOGD("Skip already persisted blob_id={}, re-committed blk_id={}", blob->blob_id(), blk_id.to_string()); skipped_blobs++; continue; } diff --git a/src/lib/homestore_backend/tests/homeobj_misc_tests.cpp b/src/lib/homestore_backend/tests/homeobj_misc_tests.cpp index 6f7a221f..c8340a3d 100644 --- a/src/lib/homestore_backend/tests/homeobj_misc_tests.cpp +++ b/src/lib/homestore_backend/tests/homeobj_misc_tests.cpp @@ -1,6 +1,7 @@ #include "homeobj_fixture.hpp" #include "generated/resync_blob_data_generated.h" #include +#include // CP related tests TEST_F(HomeObjectFixture, HSHomeObjectCPTestBasic) { @@ -360,7 +361,8 @@ TEST_F(HomeObjectFixture, PGBlobIteratorSealedLsnCutoff) { ASSERT_GT(snp_lsn, pg_shard_1->info.create_lsn); ASSERT_LT(snp_lsn, pg_shard_1->info.sealed_lsn); - auto pg_iter = std::make_shared< HSHomeObject::PGBlobIterator >(*_obj_inst, pg->pg_info_.replica_set_uuid, snp_lsn); + auto pg_iter = + std::make_shared< HSHomeObject::PGBlobIterator >(*_obj_inst, pg->pg_info_.replica_set_uuid, snp_lsn); ASSERT_EQ(pg_iter->shard_list_.size(), 2u); auto it1 = std::find_if(pg_iter->shard_list_.begin(), pg_iter->shard_list_.end(), @@ -380,7 +382,8 @@ TEST_F(HomeObjectFixture, PGBlobIteratorSealedLsnCutoff) { { auto snp_lsn = pg_shard_1->info.sealed_lsn; - auto pg_iter = std::make_shared< HSHomeObject::PGBlobIterator >(*_obj_inst, pg->pg_info_.replica_set_uuid, snp_lsn); + auto pg_iter = + std::make_shared< HSHomeObject::PGBlobIterator >(*_obj_inst, pg->pg_info_.replica_set_uuid, snp_lsn); ASSERT_EQ(pg_iter->shard_list_.size(), 2u); auto it1 = std::find_if(pg_iter->shard_list_.begin(), pg_iter->shard_list_.end(), @@ -592,3 +595,205 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandler) { } } } + +// Test: verify the case (https://jirap.corp.ebay.com/browse/SDSTOR-23023) when the snapshot receiver re-delivers a blob +// batch that was already indexed (because the IndexSvc journal survived a crash but the AppendBlkAllocator CP superblk +// did not), the allocator watermarks are correctly restored via commit_blk(recommit=true) so that subsequent +// allocations cannot silently reuse the already-occupied blocks. +// +// Scenario: +// 1. Partial first delivery (blobs 0..num_blobs_partial-1) — written, committed, indexed. +// 2. Crash simulated via VChunk::reset(), zeroing the allocator watermarks exactly as +// AppendBlkAllocator::on_meta_blk_found() would after a CP that missed the allocator flush. +// 3. Full second delivery (blobs 0..num_blobs_total-1) — blobs 0..num_blobs_partial-1 hit the +// dedup path and must restore watermarks; blobs num_blobs_partial..num_blobs_total-1 are new +// and must be allocated strictly above the already-written blocks. +// +// Without the fix the allocator watermarks stay at 0 after step 3, so the new blobs (step 3) get +// allocated from blk 0, silently overwriting the data written in step 1. +TEST_F(HomeObjectFixture, SnapshotReceiveHandlerAllocatorResyncAfterCrash) { + constexpr uint64_t snp_lsn = 1; + constexpr pg_id_t pg_id = 1; + constexpr blob_id_t num_blobs_partial = 2; // blobs written before the simulated crash + constexpr blob_id_t num_blobs_total = 4; // full set re-delivered after crash + + // ---- Setup: create PG and handler (mirrors SnapshotReceiveHandler test) ---- + create_pg(pg_id); + auto pg = _obj_inst->get_hs_pg(pg_id); + ASSERT_NE(pg, nullptr); + PGStats stats; + ASSERT_TRUE(_obj_inst->pg_manager()->get_stats(pg_id, stats)); + auto r_dev = homestore::HomeStore::instance()->repl_service().get_repl_dev(stats.replica_set_uuid); + ASSERT_TRUE(r_dev.hasValue()); + + auto handler = std::make_unique< homeobject::HSHomeObject::SnapshotReceiveHandler >(*_obj_inst, r_dev.value()); + handler->reset_context_and_metrics(snp_lsn, pg_id); + + // ---- PG meta ---- + flatbuffers::FlatBufferBuilder builder; + std::vector< flatbuffers::Offset< Member > > members; + std::vector< uint8_t > uuid(stats.replica_set_uuid.begin(), stats.replica_set_uuid.end()); + for (auto& member : stats.members) { + auto priority = member.id == stats.leader_id ? 1 : 0; + auto id = std::vector< uint8_t >(member.id.begin(), member.id.end()); + members.push_back(CreateMemberDirect(builder, &id, member.name.c_str(), priority)); + } + std::vector< uint64_t > shard_ids = {1}; + auto pg_entry = CreateResyncPGMetaDataDirect(builder, pg_id, &uuid, pg->pg_info_.size, + pg->pg_info_.expected_member_num, pg->pg_info_.chunk_size, + num_blobs_total /*blob_seq_num*/, 1 /*shard_seq_num*/, &members, + &shard_ids); + builder.Finish(pg_entry); + ASSERT_EQ(handler->process_pg_snapshot_data(*GetResyncPGMetaData(builder.GetBufferPointer())), 0); + builder.Reset(); + + // ---- Shard meta ---- + constexpr shard_id_t shard_id = 1; + auto exVchunk = _obj_inst->chunk_selector()->pick_most_available_blk_chunk(shard_id, pg_id); + ASSERT_NE(exVchunk, nullptr); + ASSERT_TRUE(exVchunk->m_v_chunk_id.has_value()); + + auto shard_entry = CreateResyncShardMetaData( + builder, shard_id, pg_id, static_cast< uint8_t >(ShardInfo::State::OPEN), snp_lsn /*create_lsn*/, + get_time_since_epoch_ms() /*created_time*/, get_time_since_epoch_ms() /*modified_time*/, 1024 * Mi /*capacity*/, + exVchunk->m_v_chunk_id.value(), 0 /*meta*/, 0 /*sealed_lsn*/); + builder.Finish(shard_entry); + ASSERT_EQ(handler->process_shard_snapshot_data(*GetResyncShardMetaData(builder.GetBufferPointer())), 0); + builder.Reset(); + + // ---- Helper: build a blob batch covering [start_blob_id, end_blob_id) ---- + auto make_blob_batch = [&](flatbuffers::FlatBufferBuilder& b, blob_id_t start_blob_id, blob_id_t end_blob_id, + bool is_last) { + std::vector< flatbuffers::Offset< ResyncBlobData > > blob_entries; + for (blob_id_t blob_id = start_blob_id; blob_id < end_blob_id; blob_id++) { + auto blob = build_blob(blob_id); + const auto aligned_hdr_size = sisl::round_up(sizeof(HSHomeObject::BlobHeader), _obj_inst->_data_block_size); + sisl::io_blob_safe blob_raw(aligned_hdr_size + blob.body.size(), io_align); + + HSHomeObject::BlobHeader hdr; + hdr.type = HSHomeObject::DataHeader::data_type_t::BLOB_INFO; + hdr.shard_id = shard_id; + hdr.blob_id = blob_id; + hdr.hash_algorithm = HSHomeObject::BlobHeader::HashAlgorithm::CRC32; + hdr.blob_size = blob.body.size(); + hdr.user_key_size = blob.user_key.size(); + hdr.object_offset = blob.object_off; + hdr.data_offset = aligned_hdr_size; + if (!blob.user_key.empty()) { std::memcpy(hdr.user_key, blob.user_key.data(), blob.user_key.size()); } + _obj_inst->compute_blob_payload_hash(hdr.hash_algorithm, blob.body.cbytes(), blob.body.size(), hdr.hash, + HSHomeObject::BlobHeader::blob_max_hash_len); + hdr.seal(); + std::memcpy(blob_raw.bytes(), &hdr, sizeof(HSHomeObject::BlobHeader)); + std::memcpy(blob_raw.bytes() + hdr.data_offset, blob.body.cbytes(), blob.body.size()); + + std::vector< uint8_t > data(blob_raw.bytes(), blob_raw.bytes() + blob_raw.size()); + blob_entries.push_back( + CreateResyncBlobDataDirect(b, blob_id, static_cast< uint8_t >(ResyncBlobState::NORMAL), &data)); + } + b.Finish(CreateResyncBlobDataBatchDirect(b, &blob_entries, is_last)); + }; + + // ---- First delivery: partial batch (blobs 0..num_blobs_partial-1), not the last batch ---- + // Simulates the pre-crash state: these blobs are written to disk, committed, and indexed. + // is_last_batch=false so update_snp_info_sb (the CP+shard-cursor write) is NOT triggered, + // matching the real crash scenario where the shard was only partially received. + LOGINFO("TEST: first (partial) delivery, blobs [0, {})", num_blobs_partial); + make_blob_batch(builder, 0, num_blobs_partial, false /*is_last*/); + ASSERT_EQ(handler->process_blobs_snapshot_data(*GetResyncBlobDataBatch(builder.GetBufferPointer()), 1 /*batch_num*/, + false /*is_last_batch*/), + 0); + builder.Reset(); + + // Confirm only the partial set is readable and indexed + for (blob_id_t blob_id = 0; blob_id < num_blobs_partial; blob_id++) { + auto result = _obj_inst->blob_manager()->get(shard_id, blob_id, 0, 0).get(); + ASSERT_TRUE(!!result) << "blob_id=" << blob_id << " should be readable after partial delivery"; + } + for (blob_id_t blob_id = num_blobs_partial; blob_id < num_blobs_total; blob_id++) { + auto result = _obj_inst->blob_manager()->get(shard_id, blob_id, 0, 0).get(); + ASSERT_FALSE(!!result) << "blob_id=" << blob_id << " should NOT exist before full delivery"; + } + + // Capture the blk_ids assigned to the partial set — these are the blocks the allocator must + // protect during re-delivery so new blobs don't overwrite them. + auto index_table = _obj_inst->get_hs_pg(pg_id)->index_table_; + ASSERT_NE(index_table, nullptr); + std::vector< homestore::MultiBlkId > partial_blk_ids; + for (blob_id_t blob_id = 0; blob_id < num_blobs_partial; blob_id++) { + auto res = _obj_inst->get_blob_from_index_table(index_table, shard_id, blob_id); + ASSERT_TRUE(res.hasValue()) << "blob_id=" << blob_id << " must be in index after partial delivery"; + partial_blk_ids.push_back(res.value()); + } + + // ---- Simulate crash: reset allocator watermarks to zero ---- + // Replicates AppendBlkAllocator::on_meta_blk_found() recovering from a CP that captured + // the IndexSvc BTree entries for blobs 0..num_blobs_partial-1 but did NOT flush the + // AppendBlkAllocator commit_offset superblock. Both watermarks (m_last_append_offset, + // m_commit_offset) are restored to 0, so the allocator considers the chunk fully empty. + auto p_chunk_id = _obj_inst->get_shard_p_chunk_id(shard_id); + ASSERT_TRUE(p_chunk_id.has_value()); + auto vchunk = _obj_inst->chunk_selector()->m_chunks.at(*p_chunk_id); + ASSERT_NE(vchunk, nullptr); + + const auto used_blks_after_partial = vchunk->get_used_blks(); + ASSERT_GT(used_blks_after_partial, 0u) << "allocator must have advanced after partial delivery"; + + LOGINFO("TEST: simulating crash — resetting allocator watermarks (used_blks was {})", used_blks_after_partial); + vchunk->reset(); // zeroes m_last_append_offset and m_commit_offset + ASSERT_EQ(vchunk->get_used_blks(), 0u) << "used_blks must be 0 after reset (simulating crashed CP)"; + + // ---- Second delivery: full batch (blobs 0..num_blobs_total-1), is_last_batch=true ---- + // Blobs 0..num_blobs_partial-1 are already in the index → dedup path fires. + // Without fix: watermarks stay at 0; blobs num_blobs_partial..num_blobs_total-1 get + // allocated from blk 0, silently overwriting the data for blobs 0..num_blobs_partial-1. + // With fix: commit_blk(recommit=true) restores watermarks to cover the partial set; + // new blobs are then allocated strictly above them. + // Blobs num_blobs_partial..num_blobs_total-1 are new → normal write path. + LOGINFO("TEST: second (full) delivery after crash, blobs [0, {})", num_blobs_total); + handler->ctx_->cur_batch_num = 0; // reset so batch_num=1 is accepted again + make_blob_batch(builder, 0, num_blobs_total, true /*is_last*/); + ASSERT_EQ(handler->process_blobs_snapshot_data(*GetResyncBlobDataBatch(builder.GetBufferPointer()), 1 /*batch_num*/, + true /*is_last_batch*/), + 0); + builder.Reset(); + + // ---- Verify allocator watermarks advanced beyond the partial set ---- + // The new blobs (num_blobs_partial..num_blobs_total-1) must have been allocated above the + // partial set, so used_blks must now exceed what the partial delivery alone had consumed. + const auto used_blks_after_full = vchunk->get_used_blks(); + LOGINFO("TEST: used_blks after full delivery: {} (partial delivery had {})", used_blks_after_full, + used_blks_after_partial); + ASSERT_GT(used_blks_after_full, used_blks_after_partial) + << "new blobs must have been allocated above the partial set, advancing the watermark further"; + + // ---- Verify no overlap: a fresh alloc must land ABOVE all blobs (partial + new) ---- + // Core safety check: without the fix, alloc_blks() returns blk 0 (watermark stayed 0), + // overlapping the data written for blobs 0..num_blobs_partial-1 in the first delivery. + homestore::MultiBlkId new_blk_id; + homestore::blk_alloc_hints hints; + hints.chunk_id_hint = *p_chunk_id; + auto alloc_status = + homestore::data_service().alloc_blks(homestore::data_service().get_blk_size(), hints, new_blk_id); + ASSERT_EQ(alloc_status, homestore::BlkAllocStatus::SUCCESS); + + for (const auto& existing_blk : partial_blk_ids) { + auto existing_end = existing_blk.blk_num() + existing_blk.blk_count(); + ASSERT_GE(new_blk_id.blk_num(), existing_end) + << "new allocation at blk " << new_blk_id.blk_num() << " must not overlap partial blob at blks [" + << existing_blk.blk_num() << ", " << existing_end << ")"; + } + homestore::data_service().async_free_blk(new_blk_id).get(); + + // ---- Verify all blobs are readable with correct content ---- + // Blobs 0..num_blobs_partial-1: must retain the data from the first delivery (not overwritten + // by the new-blob allocations in the second delivery). + // Blobs num_blobs_partial..num_blobs_total-1: must be readable as newly written. + for (blob_id_t blob_id = 0; blob_id < num_blobs_total; blob_id++) { + auto result = _obj_inst->blob_manager()->get(shard_id, blob_id, 0, 0).get(); + ASSERT_TRUE(!!result) << "blob_id=" << blob_id << " must be readable after full delivery"; + auto expected = build_blob(blob_id); + ASSERT_EQ(result->body.size(), expected.body.size()); + ASSERT_EQ(std::memcmp(result->body.bytes(), expected.body.cbytes(), expected.body.size()), 0) + << "blob_id=" << blob_id << " data must match expected content"; + } +}