SDSTOR-25274: improve stability of scrubber tests

conanfile.py

@@ -10,7 +10,7 @@
 
 class HomeObjectConan(ConanFile):
     name = "homeobject"
-    version = "4.3.1"
+    version = "4.3.2"
 
     homepage = "https://github.com/eBay/HomeObject"
     description = "Blob Store built on HomeStore"

src/lib/homestore_backend/CMakeLists.txt

@@ -84,8 +84,7 @@ target_sources(homestore_test_misc PRIVATE $<TARGET_OBJECTS:homestore_tests_misc
 target_link_libraries(homestore_test_misc PUBLIC homeobject_homestore ${COMMON_TEST_DEPS})
 add_test(NAME HomestoreTestMisc COMMAND homestore_test_misc --executor immediate --config_path ./
         --override_config homestore_config.consensus.snapshot_freq_distance=0
-        --override_config homestore_config.consensus.max_grpc_message_size=138412032
-        --override_config hs_backend_config.enable_gc=false)
+        --override_config homestore_config.consensus.max_grpc_message_size=138412032)
 
 # Dynamic tests
 add_executable(homestore_test_dynamic)
@@ -159,7 +158,6 @@ add_test(NAME HomestoreResyncTestWithLeaderRestart
 # GC tests
 add_test(NAME FetchDataWithOriginatorGC
         COMMAND homestore_test_dynamic --executor immediate --config_path ./
-        --override_config hs_backend_config.enable_gc=true
         --override_config hs_backend_config.gc_garbage_rate_threshold=50
         --override_config hs_backend_config.gc_garbage_rate_threshold_low=30
         --gtest_filter=HomeObjectFixture.FetchDataWithOriginatorGC)
@@ -168,7 +166,6 @@ add_executable(homestore_test_gc)
 target_sources(homestore_test_gc PRIVATE $<TARGET_OBJECTS:homestore_tests_gc>)
 target_link_libraries(homestore_test_gc PUBLIC homeobject_homestore ${COMMON_TEST_DEPS})
 add_test(NAME HomestoreTestGC COMMAND homestore_test_gc --executor immediate --config_path ./
-        --override_config hs_backend_config.enable_gc=true
         --override_config hs_backend_config.gc_garbage_rate_threshold=0
         --override_config hs_backend_config.gc_garbage_rate_threshold_low=0
         --override_config hs_backend_config.gc_scan_interval_sec=5
@@ -179,7 +176,6 @@ add_test(NAME HomestoreTestGC COMMAND homestore_test_gc --executor immediate --c
 add_test(NAME HomestoreTestGC_ShardRaceTests
         COMMAND homestore_test_gc --executor immediate --config_path ./
         --chunks_per_pg 1
-        --override_config hs_backend_config.enable_gc=true
         --override_config hs_backend_config.gc_garbage_rate_threshold=0
         --override_config hs_backend_config.gc_garbage_rate_threshold_low=0
         --override_config hs_backend_config.gc_scan_interval_sec=5

src/lib/homestore_backend/gc_manager.cpp

@@ -709,12 +709,12 @@ bool GCManager::pdev_gc_actor::replace_blob_index(
         // 1 if the key exist, and the filter returns homestore::put_filter_decision::replace, the ret will be
         // homestore::btree_status_t::success
 
-        // 2 if the key exist , and the filter returns homestore::put_filter_decision::remove,  the ret will be
+        // 2 if the key exist , and the filter returns homestore::put_filter_decision::remove, the ret will be
         // homestore::btree_status_t::filtered_out.(this might happen if a key is deleted after data copy but before
         // replace index)
 
-        // 3 if the key does not exist, the ret will be homestore::btree_status_t::not_found(this might
-        // happen when crash recovery)
+        // 3 if the key does not exist, the ret will be homestore::btree_status_t::not_found(this might happen when
+        // crash recovery)
 
         if (ret != homestore::btree_status_t::success && ret != homestore::btree_status_t::filtered_out &&
             ret != homestore::btree_status_t::not_found) {
@@ -1364,8 +1364,14 @@ bool GCManager::pdev_gc_actor::process_after_gc_metablk_persisted(
     // now, all the blob indexes have been replaced successfully, we can destroy the gc task superblk
     gc_task_sb.destroy();
 
-    const auto reclaimed_blk_count = m_chunk_selector->get_extend_vchunk(move_from_chunk)->get_used_blks() -
-        m_chunk_selector->get_extend_vchunk(move_to_chunk)->get_used_blks();
+    const auto used_blks_in_move_from_chunk = m_chunk_selector->get_extend_vchunk(move_from_chunk)->get_used_blks();
+    const auto used_blks_in_move_to_chunk = m_chunk_selector->get_extend_vchunk(move_to_chunk)->get_used_blks();
+
+    RELEASE_ASSERT(used_blks_in_move_from_chunk >= used_blks_in_move_to_chunk,
+                   "used blks in move_from_chunk={} should be greater than or equal to used blks in move_to_chunk={}",
+                   move_from_chunk, move_to_chunk);
+
+    const auto reclaimed_blk_count = used_blks_in_move_from_chunk - used_blks_in_move_to_chunk;
 
     durable_entities_update([this, priority, reclaimed_blk_count](auto& de) {
         priority == static_cast< uint8_t >(task_priority::normal)

src/lib/homestore_backend/hs_shard_manager.cpp

@@ -745,8 +745,6 @@ const std::set< shard_id_t > HSHomeObject::get_shards_in_chunk(homestore::chunk_
 
 void HSHomeObject::update_shard_meta_after_gc(const homestore::chunk_num_t move_from_chunk,
                                               const homestore::chunk_num_t move_to_chunk, const uint64_t task_id) {
-    auto shards = get_shards_in_chunk(move_from_chunk);
-
     // TODO::optimize this lock
     std::scoped_lock lock_guard(_shard_lock);
 

src/lib/homestore_backend/pg_blob_iterator.cpp

@@ -123,8 +123,8 @@ bool HSHomeObject::PGBlobIterator::update_cursor(const objId& id) {
         cur_batch_blob_count_ = 0;
     }
     cur_obj_id = id;
-    LOGD("Advanced resync cursor: pg={}, obj={}, shard_index={}, blob_index={}", pg_id, id.to_string(),
-         cur_shard_idx_, cur_start_blob_idx_);
+    LOGD("Advanced resync cursor: pg={}, obj={}, shard_index={}, blob_index={}", pg_id, id.to_string(), cur_shard_idx_,
+         cur_start_blob_idx_);
     return true;
 }
 
@@ -215,8 +215,8 @@ bool HSHomeObject::PGBlobIterator::create_pg_snapshot_data(sisl::io_blob_safe& m
     builder_.FinishSizePrefixed(pg_entry);
 
     pack_resync_message(meta_blob, SyncMessageType::PG_META);
-    LOGI("Created resync PG metadata: pg={}, shards={}, active_blobs={}, occupied_bytes={}", pg_id,
-         shard_ids.size(), total_blobs, total_bytes);
+    LOGI("Created resync PG metadata: pg={}, shards={}, active_blobs={}, occupied_bytes={}", pg_id, shard_ids.size(),
+         total_blobs, total_bytes);
     return true;
 }
 
@@ -236,8 +236,8 @@ bool HSHomeObject::PGBlobIterator::generate_shard_blob_list() {
 #endif
     auto r = home_obj_.query_blobs_in_shard(pg_id, cur_obj_id.shard_seq_num, 0, UINT64_MAX);
     if (!r) {
-        LOGE("Failed to query resync shard blobs: pg={}, shard_seq=0x{:x}, error={}", pg_id,
-             cur_obj_id.shard_seq_num, r.error());
+        LOGE("Failed to query resync shard blobs: pg={}, shard_seq=0x{:x}, error={}", pg_id, cur_obj_id.shard_seq_num,
+             r.error());
         return false;
     }
     cur_blob_list_ = r.value();
@@ -284,9 +284,8 @@ HSHomeObject::PGBlobIterator::load_blob_data_with_blkid(shard_id_t shard_id, blo
     sgs.size = total_size;
     sgs.iovs.emplace_back(iovec{.iov_base = read_buf.bytes(), .iov_len = read_buf.size()});
 
-    LOGT("Reading resync blob: pg={}, shard=0x{:x}, blob={}, blkid={}, bytes={}",
-         (shard_id >> homeobject::shard_width), (shard_id & homeobject::shard_mask), blob_id, blkid.to_string(),
-         total_size);
+    LOGT("Reading resync blob: pg={}, shard=0x{:x}, blob={}, blkid={}, bytes={}", (shard_id >> homeobject::shard_width),
+         (shard_id & homeobject::shard_mask), blob_id, blkid.to_string(), total_size);
     return repl_dev_->async_read(blkid, sgs, total_size)
         .thenValue([this, blob_id, shard_id, blkid, read_buf = std::move(read_buf)](
                        auto&& result) mutable -> BlobManager::AsyncResult< blob_read_result > {
@@ -316,8 +315,8 @@ HSHomeObject::PGBlobIterator::load_blob_data_with_blkid(shard_id_t shard_id, blo
                 // Blob was deleted concurrently after generate_shard_blob_list captured its pbas.
                 // Do not send stale bytes as CORRUPTED — signal READ_FAILED so the snapshot restarts
                 // and generate_shard_blob_list picks up tombstone_pbas, skipping the blob cleanly.
-                LOGW("Resync blob was deleted during read; restarting snapshot: pg={}, shard_id=0x{:x}, blob={}",
-                     pg_id, shard_id, blob_id);
+                LOGW("Resync blob was deleted during read; restarting snapshot: pg={}, shard_id=0x{:x}, blob={}", pg_id,
+                     shard_id, blob_id);
                 return folly::makeUnexpected(BlobError(BlobErrorCode::READ_FAILED));
             }
             if (current_pbas.value() == blkid) {
@@ -330,7 +329,8 @@ HSHomeObject::PGBlobIterator::load_blob_data_with_blkid(shard_id_t shard_id, blo
             }
 
             // GC moved the blob — retry with the updated blkid. Folly flattens the returned future.
-            LOGI("Resync blob relocated by GC during read; retrying: pg={}, shard_id=0x{:x}, blob={}, old_blkid={}, new_blkid={}",
+            LOGI("Resync blob relocated by GC during read; retrying: pg={}, shard_id=0x{:x}, blob={}, old_blkid={}, "
+                 "new_blkid={}",
                  pg_id, shard_id, blob_id, blkid.to_string(), current_pbas.value().to_string());
             return load_blob_data_with_blkid(shard_id, blob_id, current_pbas.value());
         });
@@ -345,8 +345,8 @@ bool HSHomeObject::PGBlobIterator::prefetch_blobs_snapshot_data() {
     // On batch resend, retained look-ahead may already consume part of the 2x budget. Allow missing blobs before the
     // earliest retained blob to bypass the limit so the current batch can always be rebuilt.
     const auto prefetch_frontier = prefetched_blobs_.empty() ? blob_id_t{0} : prefetched_blobs_.begin()->first;
-    LOGT("Prefetching blobs: pg={}, shard_seq=0x{:x}, cursor_blob={}, frontier={}, inflight_bytes={}",
-         pg_id, cur_obj_id.shard_seq_num, cur_start_blob_idx_, prefetch_frontier, inflight_prefetch_bytes_);
+    LOGT("Prefetching blobs: pg={}, shard_seq=0x{:x}, cursor_blob={}, frontier={}, inflight_bytes={}", pg_id,
+         cur_obj_id.shard_seq_num, cur_start_blob_idx_, prefetch_frontier, inflight_prefetch_bytes_);
     while (idx < cur_blob_list_.size() &&
            (inflight_prefetch_bytes_ < max_batch_size_ * 2 || cur_blob_list_[idx].blob_id < prefetch_frontier)) {
         auto info = cur_blob_list_[idx++];
@@ -407,7 +407,8 @@ bool HSHomeObject::PGBlobIterator::prefetch_blobs_snapshot_data() {
                         return result;
                     }));
     }
-    LOGD("Resync prefetch window: pg={}, shard_seq=0x{:x}, cursor_blob={}, frontier={}, submitted_blobs={}, skipped_blobs={}, inflight_bytes={}, limit_bytes={}",
+    LOGD("Resync prefetch window: pg={}, shard_seq=0x{:x}, cursor_blob={}, frontier={}, submitted_blobs={}, "
+         "skipped_blobs={}, inflight_bytes={}, limit_bytes={}",
          pg_id, cur_obj_id.shard_seq_num, cur_start_blob_idx_, prefetch_frontier, prefetch_list.size(), skipped_blobs,
          inflight_prefetch_bytes_, max_batch_size_ * 2);
     return true;
@@ -440,7 +441,7 @@ bool HSHomeObject::PGBlobIterator::create_blobs_snapshot_data(sisl::io_blob_safe
             // handle deleted object
             if (info.pbas == tombstone_pbas) {
                 LOGT("Skipping deleted resync blob: pg={}, shard=0x{:x}, blob={}",
-                    info.shard_id >> homeobject::shard_width, info.shard_id & homeobject::shard_mask, info.blob_id);
+                     info.shard_id >> homeobject::shard_width, info.shard_id & homeobject::shard_mask, info.blob_id);
                 // ignore
                 skipped_blobs++;
                 continue;
@@ -449,7 +450,8 @@ bool HSHomeObject::PGBlobIterator::create_blobs_snapshot_data(sisl::io_blob_safe
             auto it = prefetched_blobs_.find(info.blob_id);
             if (it == prefetched_blobs_.end()) {
                 hit_error = true;
-                LOGE("Resync batch cannot find prefetched blob: pg={}, shard_seq=0x{:x}, batch={}, blob={}, cursor_blob={}, inflight_bytes={}, prefetched_blobs={}",
+                LOGE("Resync batch cannot find prefetched blob: pg={}, shard_seq=0x{:x}, batch={}, blob={}, "
+                     "cursor_blob={}, inflight_bytes={}, prefetched_blobs={}",
                      pg_id, cur_obj_id.shard_seq_num, cur_obj_id.batch_id, info.blob_id, cur_start_blob_idx_,
                      inflight_prefetch_bytes_, prefetched_blobs_.size());
                 break;
@@ -473,7 +475,8 @@ bool HSHomeObject::PGBlobIterator::create_blobs_snapshot_data(sisl::io_blob_safe
     }
 
     if (skipped_blobs + fetched_blobs != total_blobs) {
-        LOGE("Incomplete resync batch: pg={}, shard_seq=0x{:x}, batch={}, examined_blobs={}, skipped_blobs={}, expected_blobs={}, fetched_blobs={}",
+        LOGE("Incomplete resync batch: pg={}, shard_seq=0x{:x}, batch={}, examined_blobs={}, skipped_blobs={}, "
+             "expected_blobs={}, fetched_blobs={}",
              pg_id, cur_obj_id.shard_seq_num, cur_obj_id.batch_id, total_blobs, skipped_blobs,
              total_blobs - skipped_blobs, fetched_blobs);
         hit_error = true;
@@ -489,7 +492,8 @@ bool HSHomeObject::PGBlobIterator::create_blobs_snapshot_data(sisl::io_blob_safe
     if (idx == cur_blob_list_.size()) { end_of_shard = true; }
     builder_.FinishSizePrefixed(CreateResyncBlobDataBatchDirect(builder_, &blob_entries, end_of_shard));
 
-    LOGI("Created resync shard batch: pg={}, shard_seq=0x{:x}, batch={}, blobs={}, skipped_blobs={}, bytes={}, end_of_shard={}, next_blob={}",
+    LOGI("Created resync shard batch: pg={}, shard_seq=0x{:x}, batch={}, blobs={}, skipped_blobs={}, bytes={}, "
+         "end_of_shard={}, next_blob={}",
          pg_id, cur_obj_id.shard_seq_num, cur_obj_id.batch_id, blob_entries.size(), skipped_blobs, total_bytes,
          end_of_shard, idx);
 

src/lib/homestore_backend/replication_state_machine.cpp

@@ -404,8 +404,8 @@ void ReplicationStateMachine::write_snapshot_obj(std::shared_ptr< homestore::sna
         m_snp_rcv_handler = std::make_unique< HSHomeObject::SnapshotReceiveHandler >(*home_object_, r_dev);
         if (m_snp_rcv_handler->load_prev_context_and_metrics()) {
             LOGI("Reloaded resync receiver context: lsn={}, pg={}, next_shard=0x{:x}",
-                 m_snp_rcv_handler->get_context_lsn(),
-                 m_snp_rcv_handler->get_context_pg_id(), m_snp_rcv_handler->get_next_shard());
+                 m_snp_rcv_handler->get_context_lsn(), m_snp_rcv_handler->get_context_pg_id(),
+                 m_snp_rcv_handler->get_next_shard());
         }
     }
 
@@ -476,8 +476,8 @@ void ReplicationStateMachine::write_snapshot_obj(std::shared_ptr< homestore::sna
                 return;
             }
         }
-        LOGD("Resetting resync receiver context: previous_lsn={}, new_lsn={}",
-             m_snp_rcv_handler->get_context_lsn(), context->get_lsn());
+        LOGD("Resetting resync receiver context: previous_lsn={}, new_lsn={}", m_snp_rcv_handler->get_context_lsn(),
+             context->get_lsn());
         m_snp_rcv_handler->reset_context_and_metrics(context->get_lsn(), pg_data->pg_id());
 
         auto ret = m_snp_rcv_handler->process_pg_snapshot_data(*pg_data);

src/lib/homestore_backend/scrub_manager.cpp

@@ -840,10 +840,14 @@ void ScrubManager::handle_pg_scrub_task(scrub_task task) {
         scrub_task& task;
         std::shared_ptr< ShallowScrubReport >& scrub_report;
         const pg_id_t& pg_id;
+        std::shared_ptr< PGScrubContext > scrub_ctx; // set after ctx is emplaced; nullptr until then
 
         ~scrub_task_guard() {
+            // Return nullptr when cancelled so callers can distinguish cancellation from a
+            // completed scrub that legitimately found no issues (e.g. an empty PG).
+            const bool was_cancelled = scrub_ctx && scrub_ctx->cancelled.load();
             pg_scrub_ctx_map.erase(pg_id);
-            task.scrub_report_promise->setValue(scrub_report);
+            task.scrub_report_promise->setValue(was_cancelled ? nullptr : scrub_report);
             auto hs_pg = home_obj->get_hs_pg(pg_id);
             if (hs_pg) {
                 hs_pg->in_scrubbing.store(false);
@@ -853,7 +857,7 @@ void ScrubManager::handle_pg_scrub_task(scrub_task task) {
                 LOGWARNMOD(scrubmgr, "cannot find hs_pg to clear SCRUBBING state for pg={}!", pg_id);
             }
         }
-    } guard{m_hs_home_object, m_pg_scrub_ctx_map, task, pg_scrub_report, pg_id};
+    } guard{m_hs_home_object, m_pg_scrub_ctx_map, task, pg_scrub_report, pg_id, nullptr};
 
     const auto hs_pg = m_hs_home_object->get_hs_pg(pg_id);
     if (!hs_pg) {
@@ -865,6 +869,7 @@ void ScrubManager::handle_pg_scrub_task(scrub_task task) {
     RELEASE_ASSERT(happened,
                    "pg={} should not have a running scrub task since we set in_scrubbing in submit_scrub_task", pg_id);
     auto& scrub_ctx = ctx_it->second;
+    guard.scrub_ctx = scrub_ctx; // Allow the guard to detect cancellation at teardown
 
     // this is the last committed shard_id. we cannot get shard_sequence_num here since some of the shard might be
     // not committed yet. note that, this depends on the fact that the last committed shard is always at the end of

src/lib/homestore_backend/tests/homeobj_fixture.hpp

@@ -776,6 +776,29 @@ class HomeObjectFixture : public ::testing::Test {
         // TODO: add logic for check and retry of leader change if necessary
     }
 
+    // Submit a scrub task and wait for it; retry if the task was cancelled mid-flight (e.g. due
+    // to a leader switch).  Returns nullptr on non-leader replicas.  RELEASE_ASSERTs if the
+    // retry deadline is exceeded while this replica is still the leader.
+    std::shared_ptr< ScrubManager::ShallowScrubReport > submit_scrub_with_retry(pg_id_t pg_id, bool is_deep,
+                                                                                uint32_t timeout_secs = 60) {
+        auto scrub_mgr = _obj_inst->scrub_manager();
+        auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(timeout_secs);
+        while (std::chrono::steady_clock::now() < deadline) {
+            PGStats pg_stats;
+            if (!_obj_inst->pg_manager()->get_stats(pg_id, pg_stats)) return nullptr;
+            if (g_helper->my_replica_id() != pg_stats.leader_id) return nullptr;
+
+            auto report = scrub_mgr->submit_scrub_task(pg_id, is_deep, SCRUB_TRIGGER_TYPE::MANUALLY).get();
+            if (report) return report;
+
+            // null means the task was cancelled (leader switch); re-check leadership and retry.
+            LOGWARN("scrub task cancelled for pg={} (leader switch?), retrying…", pg_id);
+            std::this_thread::sleep_for(std::chrono::milliseconds(200));
+        }
+        RELEASE_ASSERT(false, "submit_scrub_with_retry: timeout after {}s for pg={}", timeout_secs, pg_id);
+        return nullptr;
+    }
+
     void run_on_pg_follower(pg_id_t pg_id, auto&& lambda) {
         PGStats pg_stats;
         auto res = _obj_inst->pg_manager()->get_stats(pg_id, pg_stats);