diff --git a/conanfile.py b/conanfile.py index 1840dbe7..4b6ddf57 100644 --- a/conanfile.py +++ b/conanfile.py @@ -10,7 +10,7 @@ class HomeObjectConan(ConanFile): name = "homeobject" - version = "4.2.6" + version = "4.2.7" homepage = "https://github.com/eBay/HomeObject" description = "Blob Store built on HomeStore" diff --git a/src/lib/homestore_backend/hs_homeobject.hpp b/src/lib/homestore_backend/hs_homeobject.hpp index d5c7cf65..faeb2e0c 100644 --- a/src/lib/homestore_backend/hs_homeobject.hpp +++ b/src/lib/homestore_backend/hs_homeobject.hpp @@ -667,8 +667,14 @@ class HSHomeObject : public HomeObjectImpl { const int64_t snp_lsn; const pg_id_t pg_id; shared< BlobIndexTable > index_table; + + // Progress and accounting fields + uint64_t cur_shard_complete_blobs{0}; + uint64_t cur_shard_complete_bytes{0}; + bool cur_shard_completed{false}; std::shared_mutex progress_lock; snapshot_progress progress; + SnapshotContext(int64_t lsn, pg_id_t pg_id) : snp_lsn{lsn}, pg_id{pg_id} {} }; diff --git a/src/lib/homestore_backend/snapshot_receive_handler.cpp b/src/lib/homestore_backend/snapshot_receive_handler.cpp index eacfb264..a4898a07 100644 --- a/src/lib/homestore_backend/snapshot_receive_handler.cpp +++ b/src/lib/homestore_backend/snapshot_receive_handler.cpp @@ -92,6 +92,19 @@ int HSHomeObject::SnapshotReceiveHandler::process_shard_snapshot_data(ResyncShar shard_sb->v_chunk_id = shard_meta.vchunk_id(); // Now let's create local shard home_obj_.local_create_shard(shard_sb->info, shard_sb->v_chunk_id); + { + std::unique_lock< std::shared_mutex > lock(ctx_->progress_lock); + if (ctx_->shard_cursor == shard_meta.shard_id()) { + ctx_->progress.complete_blobs -= ctx_->cur_shard_complete_blobs; + ctx_->progress.complete_bytes -= ctx_->cur_shard_complete_bytes; + if (ctx_->cur_shard_completed) { ctx_->progress.complete_shards--; } + } + ctx_->cur_shard_complete_blobs = 0; + ctx_->cur_shard_complete_bytes = 0; + ctx_->cur_shard_completed = false; + ctx_->progress.cur_batch_blobs = 0; + ctx_->progress.cur_batch_bytes = 0; + } ctx_->shard_cursor = shard_meta.shard_id(); ctx_->cur_batch_num = 0; return 0; @@ -117,6 +130,13 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob std::unique_lock< std::shared_mutex > lock(ctx_->progress_lock); ctx_->progress.complete_blobs -= ctx_->progress.cur_batch_blobs; ctx_->progress.complete_bytes -= ctx_->progress.cur_batch_bytes; + ctx_->cur_shard_complete_blobs -= ctx_->progress.cur_batch_blobs; + ctx_->cur_shard_complete_bytes -= ctx_->progress.cur_batch_bytes; + ctx_->progress.cur_batch_blobs = 0; + ctx_->progress.cur_batch_bytes = 0; + } else { + std::unique_lock< std::shared_mutex > lock(ctx_->progress_lock); + // A new batch must not inherit the prior batch's rollback contribution. ctx_->progress.cur_batch_blobs = 0; ctx_->progress.cur_batch_bytes = 0; } @@ -136,6 +156,7 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob std::vector< std::shared_ptr< sisl::io_blob_safe > > data_bufs; auto skipped_blobs = 0; + bool found_unexpected_corruption = false; for (unsigned int i = 0; i < data_blobs.blob_list()->size(); i++) { const auto blob = data_blobs.blob_list()->Get(i); @@ -146,6 +167,9 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob continue; } + // Count the logical batch size even when an existing blob is re-committed during replay. + total_bytes += blob->data()->size(); + auto start = Clock::now(); #ifdef _PRERELEASE @@ -188,7 +212,8 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob LOGE("Blob verification failed for blob_id={}", blob->blob_id()); std::unique_lock< std::shared_mutex > lock(ctx_->progress_lock); ctx_->progress.error_count++; - return BLOB_DATA_CORRUPTED; + found_unexpected_corruption = true; + break; } } else { LOGW("find corrupted_blobs={} in shardID=0x{:x}, pg={}, shard=0x{:x}", blob->blob_id(), ctx_->shard_cursor, @@ -271,9 +296,9 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob LOGD("Persisted blob_id={} in {}us", blob_id, duration); return std::error_code{}; })); - total_bytes += data_size; } auto ec = collect_all_futures(futs).get(); + if (found_unexpected_corruption) { return BLOB_DATA_CORRUPTED; } // when there is a allocation failure it breaks the while loop earlier. auto all_io_submitted = (futs.size() + skipped_blobs == data_blobs.blob_list()->size()); @@ -298,8 +323,9 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob ctx_->progress.cur_batch_bytes = total_bytes; ctx_->progress.complete_blobs += ctx_->progress.cur_batch_blobs; ctx_->progress.complete_bytes += ctx_->progress.cur_batch_bytes; + ctx_->cur_shard_complete_blobs += ctx_->progress.cur_batch_blobs; + ctx_->cur_shard_complete_bytes += ctx_->progress.cur_batch_bytes; } - if (is_last_batch) { // Release chunk for sealed shard ShardInfo::State state; @@ -313,7 +339,8 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob } { std::unique_lock< std::shared_mutex > lock(ctx_->progress_lock); - ctx_->progress.complete_shards++; + if (!ctx_->cur_shard_completed) { ctx_->progress.complete_shards++; } + ctx_->cur_shard_completed = true; } // We only update the snp info superblk on completion of each shard, since resumption is also shard-level update_snp_info_sb(ctx_->shard_cursor == ctx_->shard_list.front()); diff --git a/src/lib/homestore_backend/tests/homeobj_misc_tests.cpp b/src/lib/homestore_backend/tests/homeobj_misc_tests.cpp index c8340a3d..30a9dd65 100644 --- a/src/lib/homestore_backend/tests/homeobj_misc_tests.cpp +++ b/src/lib/homestore_backend/tests/homeobj_misc_tests.cpp @@ -436,7 +436,8 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandler) { } auto pg_entry = CreateResyncPGMetaDataDirect(builder, pg_id, &uuid, pg->pg_info_.size, pg->pg_info_.expected_member_num, - pg->pg_info_.chunk_size, blob_seq_num, num_shards_per_pg, &members, &shard_ids); + pg->pg_info_.chunk_size, blob_seq_num, num_shards_per_pg, &members, &shard_ids, + blob_seq_num /* total_blobs_to_transfer */); builder.Finish(pg_entry); auto pg_meta = GetResyncPGMetaData(builder.GetBufferPointer()); auto ret = handler->process_pg_snapshot_data(*pg_meta); @@ -450,6 +451,7 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandler) { std::uniform_int_distribution<> random_bytes_dis(1, 16 * 1024); blob_id_t cur_blob_id{0}; + uint64_t expected_complete_blobs{0}; for (uint64_t i = 1; i <= num_shards_per_pg; i++) { LOGINFO("TESTING: applying meta for shard {}", i); ASSERT_TRUE(handler->is_valid_obj_id(objId(HSHomeObject::get_sequence_num_from_shard_id(i), 0))); @@ -503,11 +505,12 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandler) { for (uint64_t j = 1; j <= num_batches_per_shard; j++) { ASSERT_TRUE(handler->is_valid_obj_id(objId(HSHomeObject::get_sequence_num_from_shard_id(shard.id), j))); - // Don't test unexpected corruption on the last batch, since for simplicity we're not simulating resending + // Keep the final batch clean so each shard completes even if a prior batch needs a resend. bool is_corrupted_batch = j < num_batches_per_shard && corrupt_dis(gen) <= unexpected_corrupted_batch_percentage; LOGINFO("TESTING: applying blobs for shard {} batch {}, is_corrupted {}", shard.id, j, is_corrupted_batch); std::vector< flatbuffers::Offset< ResyncBlobData > > blob_entries; + auto const batch_start_blob_id = cur_blob_id; for (uint64_t k = 0; k < num_blobs_per_batch; k++) { auto blob_state = corrupt_dis(gen) <= corrupted_blob_percentage ? ResyncBlobState::CORRUPTED : ResyncBlobState::NORMAL; @@ -560,9 +563,46 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandler) { ret = handler->process_blobs_snapshot_data(*blob_batch, j, j == num_batches_per_shard); if (is_corrupted_batch) { ASSERT_NE(ret, 0); + builder.Reset(); + std::vector< flatbuffers::Offset< ResyncBlobData > > retry_blob_entries; + for (blob_id_t retry_blob_id = batch_start_blob_id; + retry_blob_id < batch_start_blob_id + num_blobs_per_batch; retry_blob_id++) { + auto retry_blob = build_blob(retry_blob_id); + const auto aligned_hdr_size = + sisl::round_up(sizeof(HSHomeObject::BlobHeader), _obj_inst->_data_block_size); + sisl::io_blob_safe retry_blob_raw(aligned_hdr_size + retry_blob.body.size(), io_align); + HSHomeObject::BlobHeader retry_hdr; + retry_hdr.type = HSHomeObject::DataHeader::data_type_t::BLOB_INFO; + retry_hdr.shard_id = shard.id; + retry_hdr.blob_id = retry_blob_id; + retry_hdr.hash_algorithm = HSHomeObject::BlobHeader::HashAlgorithm::CRC32; + retry_hdr.blob_size = retry_blob.body.size(); + retry_hdr.user_key_size = retry_blob.user_key.size(); + retry_hdr.object_offset = retry_blob.object_off; + retry_hdr.data_offset = aligned_hdr_size; + if (!retry_blob.user_key.empty()) { + std::memcpy(retry_hdr.user_key, retry_blob.user_key.data(), retry_blob.user_key.size()); + } + _obj_inst->compute_blob_payload_hash(retry_hdr.hash_algorithm, retry_blob.body.cbytes(), + retry_blob.body.size(), retry_hdr.hash, + HSHomeObject::BlobHeader::blob_max_hash_len); + retry_hdr.seal(); + std::memcpy(retry_blob_raw.bytes(), &retry_hdr, sizeof(HSHomeObject::BlobHeader)); + std::memcpy(retry_blob_raw.bytes() + retry_hdr.data_offset, retry_blob.body.cbytes(), + retry_blob.body.size()); + std::vector retry_data(retry_blob_raw.bytes(), retry_blob_raw.bytes() + retry_blob_raw.size()); + retry_blob_entries.push_back(CreateResyncBlobDataDirect( + builder, retry_blob_id, static_cast< uint8_t >(ResyncBlobState::NORMAL), &retry_data)); + blob_map[retry_blob_id] = std::make_tuple< Blob, bool >(std::move(retry_blob), false); + } + builder.Finish(CreateResyncBlobDataBatchDirect(builder, &retry_blob_entries, j == num_batches_per_shard)); + auto retry_blob_batch = GetResyncBlobDataBatch(builder.GetBufferPointer()); + ASSERT_EQ(handler->process_blobs_snapshot_data(*retry_blob_batch, j, j == num_batches_per_shard), 0); } else { ASSERT_EQ(ret, 0); } + expected_complete_blobs += num_blobs_per_batch; + ASSERT_EQ(handler->ctx_->progress.complete_blobs, expected_complete_blobs); builder.Reset(); ASSERT_EQ(handler->get_shard_cursor(), shard.id); ASSERT_EQ(handler->get_next_shard(), @@ -593,7 +633,10 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandler) { ASSERT_EQ(v.value(), exVchunk->m_v_chunk_id.value()); ASSERT_TRUE(_obj_inst->chunk_selector()->is_chunk_available(pg_id, v.value())); } + ASSERT_EQ(handler->ctx_->progress.complete_shards, i); } + ASSERT_EQ(handler->ctx_->progress.total_blobs, blob_seq_num); + ASSERT_EQ(handler->ctx_->progress.total_shards, num_shards_per_pg); } // Test: verify the case (https://jirap.corp.ebay.com/browse/SDSTOR-23023) when the snapshot receiver re-delivers a blob @@ -692,6 +735,14 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandlerAllocatorResyncAfterCrash) { } b.Finish(CreateResyncBlobDataBatchDirect(b, &blob_entries, is_last)); }; + auto batch_bytes = [&](blob_id_t start_blob_id, blob_id_t end_blob_id) { + uint64_t total_bytes{0}; + for (blob_id_t blob_id = start_blob_id; blob_id < end_blob_id; blob_id++) { + auto blob = build_blob(blob_id); + total_bytes += sisl::round_up(sizeof(HSHomeObject::BlobHeader), _obj_inst->_data_block_size) + blob.body.size(); + } + return total_bytes; + }; // ---- 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. @@ -702,6 +753,8 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandlerAllocatorResyncAfterCrash) { ASSERT_EQ(handler->process_blobs_snapshot_data(*GetResyncBlobDataBatch(builder.GetBufferPointer()), 1 /*batch_num*/, false /*is_last_batch*/), 0); + ASSERT_EQ(handler->ctx_->progress.complete_blobs, num_blobs_partial); + ASSERT_EQ(handler->ctx_->progress.complete_bytes, batch_bytes(0, num_blobs_partial)); builder.Reset(); // Confirm only the partial set is readable and indexed @@ -750,11 +803,23 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandlerAllocatorResyncAfterCrash) { // 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 + // Replaying shard metadata resets the shard-local contribution before batch 1 is redelivered. + auto replay_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(replay_shard_entry); + ASSERT_EQ(handler->process_shard_snapshot_data(*GetResyncShardMetaData(builder.GetBufferPointer())), 0); + ASSERT_EQ(handler->ctx_->progress.complete_blobs, 0); + ASSERT_EQ(handler->ctx_->progress.complete_bytes, 0); + builder.Reset(); 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); + ASSERT_EQ(handler->ctx_->progress.complete_blobs, num_blobs_total); + ASSERT_EQ(handler->ctx_->progress.complete_bytes, batch_bytes(0, num_blobs_total)); + ASSERT_EQ(handler->ctx_->progress.complete_shards, 1); builder.Reset(); // ---- Verify allocator watermarks advanced beyond the partial set ----