Expose data corruptted api for scrubber integration test

conanfile.py

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

src/lib/homestore_backend/CMakeLists.txt

@@ -31,6 +31,7 @@ target_sources("${PROJECT_NAME}_homestore" PRIVATE
     hs_http_manager.cpp
     gc_manager.cpp
     scrub_manager.cpp
+    hs_debug_inject.cpp
     MPMCPriorityQueue.hpp
     $<TARGET_OBJECTS:${PROJECT_NAME}_core>
 )
@@ -186,7 +187,19 @@ target_sources(homestore_test_scrubber PRIVATE $<TARGET_OBJECTS:homestore_tests_
 target_link_libraries(homestore_test_scrubber PUBLIC homeobject_homestore ${COMMON_TEST_DEPS})
 add_test(NAME HomestoreTestScrubber COMMAND homestore_test_scrubber -csv error --executor immediate --config_path ./
         --override_config hs_backend_config.enable_scrubber=true
-        --override_config nuraft_mesg_config.mesg_factory_config.data_request_deadline_secs:10)
+        --override_config nuraft_mesg_config.mesg_factory_config.data_request_deadline_secs:10
+        --gtest_filter=-HomeObjectFixture.DebugInjectCorruptionHttpTest)
+
+# DebugInjectCorruptionHttpTest needs a live HTTP server (--enable_http true), which starts/stops the
+# shared process-wide iomgr http_server in every TEST_F's SetUp/TearDown. That router does not support
+# being re-registered across multiple HSHomeObject instances within one process (the first test's route
+# bindings "win" and stay bound to its now-destroyed HSHomeObject for the rest of the run), so this test
+# runs alone, as the only test in its own process invocation.
+add_test(NAME HomestoreTestScrubberHttpDebug COMMAND homestore_test_scrubber -csv error --executor immediate
+        --config_path ./ --enable_http true
+        --override_config hs_backend_config.enable_scrubber=true
+        --override_config nuraft_mesg_config.mesg_factory_config.data_request_deadline_secs:10
+        --gtest_filter=HomeObjectFixture.DebugInjectCorruptionHttpTest)
 
 # All tests share disk, port, and shared-memory resources, so run them serially to avoid parallel conflicts.
 set_tests_properties(
@@ -201,6 +214,7 @@ set_tests_properties(
     HomestoreTestGC
     HomestoreTestGC_ShardRaceTests
     HomestoreTestScrubber
+    HomestoreTestScrubberHttpDebug
     PROPERTIES RUN_SERIAL TRUE)
 
 

src/lib/homestore_backend/gc_manager.cpp

@@ -13,8 +13,8 @@ SISL_LOGGING_DECL(gcmgr)
 #define RECOVERD_GC_TASK_ID 0
 
 #define GCLOG(level, gc_task_id, pg_id, shard_id, msg, ...)                                                            \
-    LOG##level##MOD(gcmgr, "[gc_task_id={}, pg_id={}, shard_id=0x{:x}] " msg, gc_task_id, pg_id, shard_id,             \
-                    ##__VA_ARGS__)
+    LOG##level##MOD(gcmgr, "[gc_task_id={}, pg_id={}, shard_id=0x{:x}] " msg, gc_task_id, pg_id,                       \
+                    shard_id, ##__VA_ARGS__)
 
 #define GCLOGT(gc_task_id, pg_id, shard_id, msg, ...) GCLOG(TRACE, gc_task_id, pg_id, shard_id, msg, ##__VA_ARGS__)
 #define GCLOGD(gc_task_id, pg_id, shard_id, msg, ...) GCLOG(DEBUG, gc_task_id, pg_id, shard_id, msg, ##__VA_ARGS__)

src/lib/homestore_backend/gc_manager.hpp

@@ -155,14 +155,10 @@ class GCManager {
 
                 // Backlog / pressure snapshot gauges. Values refreshed once per scan cycle by
                 // GCManager::scan_chunks_for_gc; worst-case staleness = gc_scan_interval_sec.
-                REGISTER_GAUGE(pending_gc_bytes,
-                               "Total reclaimable garbage bytes in PG-owned chunks on this pdev");
-                REGISTER_GAUGE(eligible_gc_bytes,
-                               "Reclaimable bytes currently eligible for normal GC on this pdev");
-                REGISTER_GAUGE(eligible_gc_chunk_count,
-                               "Chunks currently eligible for normal GC on this pdev");
-                REGISTER_GAUGE(pending_normal_gc_task_count,
-                               "Normal-priority GC tasks queued or running on this pdev");
+                REGISTER_GAUGE(pending_gc_bytes, "Total reclaimable garbage bytes in PG-owned chunks on this pdev");
+                REGISTER_GAUGE(eligible_gc_bytes, "Reclaimable bytes currently eligible for normal GC on this pdev");
+                REGISTER_GAUGE(eligible_gc_chunk_count, "Chunks currently eligible for normal GC on this pdev");
+                REGISTER_GAUGE(pending_normal_gc_task_count, "Normal-priority GC tasks queued or running on this pdev");
 
                 // Distribution of the pending backlog by garbage-ratio bucket. We register 10
                 // gauges under a single Prometheus metric name (`pending_gc_chunks_ratio`),
@@ -179,18 +175,16 @@ class GCManager {
                 // shape. Prometheus text output is unaffected — it uses HELP (unchanged across
                 // registrations) and disambiguates series by labels.
                 static constexpr std::array< const char*, 10 > kRatioBucketLabels = {
-                    "00-10", "10-20", "20-30", "30-40", "40-50",
-                    "50-60", "60-70", "70-80", "80-90", "90-100"};
+                    "00-10", "10-20", "20-30", "30-40", "40-50", "50-60", "60-70", "70-80", "80-90", "90-100"};
                 for (size_t i = 0; i < kRatioBucketLabels.size(); ++i) {
                     const auto lo = i * 10;
                     const auto hi = (i + 1) * 10;
-                    const auto desc = fmt::format(
-                        "Snapshot count of pending chunks with garbage ratio in ({}, {}]% "
-                        "(bucket={})",
-                        lo, hi, kRatioBucketLabels[i]);
-                    ratio_bucket_indices_[i] = m_impl_ptr->register_gauge(
-                        "pending_gc_chunks_ratio", desc, "" /* report_name */,
-                        sisl::metric_label{"bucket", kRatioBucketLabels[i]});
+                    const auto desc = fmt::format("Snapshot count of pending chunks with garbage ratio in ({}, {}]% "
+                                                  "(bucket={})",
+                                                  lo, hi, kRatioBucketLabels[i]);
+                    ratio_bucket_indices_[i] =
+                        m_impl_ptr->register_gauge("pending_gc_chunks_ratio", desc, "" /* report_name */,
+                                                   sisl::metric_label{"bucket", kRatioBucketLabels[i]});
                 }
 
                 register_me_to_farm();
@@ -230,9 +224,8 @@ class GCManager {
                 // Bypass GAUGE_UPDATE for the same reason as bucket registration: we need to
                 // address 10 distinct gauge indices that share one metric name.
                 for (size_t i = 0; i < ratio_bucket_indices_.size(); ++i) {
-                    m_impl_ptr->gauge_update(
-                        ratio_bucket_indices_[i],
-                        static_cast< int64_t >(gc_actor_.get_pending_ratio_bucket(i)));
+                    m_impl_ptr->gauge_update(ratio_bucket_indices_[i],
+                                             static_cast< int64_t >(gc_actor_.get_pending_ratio_bucket(i)));
                 }
             }
 
@@ -315,12 +308,8 @@ class GCManager {
 
         // Snapshot readers used by pdev_gc_metrics::on_gather. Return the last value published
         // by GCManager::scan_chunks_for_gc for this pdev; 0 before the first scan completes.
-        uint64_t get_pending_gc_bytes() const {
-            return m_pending_gc_bytes.load(std::memory_order_relaxed);
-        }
-        uint64_t get_eligible_gc_bytes() const {
-            return m_eligible_gc_bytes.load(std::memory_order_relaxed);
-        }
+        uint64_t get_pending_gc_bytes() const { return m_pending_gc_bytes.load(std::memory_order_relaxed); }
+        uint64_t get_eligible_gc_bytes() const { return m_eligible_gc_bytes.load(std::memory_order_relaxed); }
         uint32_t get_eligible_gc_chunk_count() const {
             return m_eligible_gc_chunk_count.load(std::memory_order_relaxed);
         }
@@ -332,8 +321,7 @@ class GCManager {
         // the totals locally over all chunks on this pdev, then hand them in via one call so the
         // metrics stay internally consistent within a scan cycle. Between-gauge drift is bounded
         // by one scan interval; individual scalars are aligned and therefore torn-read safe.
-        void publish_scan_snapshot(uint64_t pending_bytes, uint64_t eligible_bytes,
-                                   uint32_t eligible_chunks,
+        void publish_scan_snapshot(uint64_t pending_bytes, uint64_t eligible_bytes, uint32_t eligible_chunks,
                                    const std::array< uint32_t, 10 >& ratio_buckets) {
             m_pending_gc_bytes.store(pending_bytes, std::memory_order_relaxed);
             m_eligible_gc_bytes.store(eligible_bytes, std::memory_order_relaxed);

src/lib/homestore_backend/hs_blob_manager.cpp

@@ -14,8 +14,8 @@ SISL_LOGGING_DECL(blobmgr)
 
 #define BLOG(level, trace_id, shard_id, blob_id, msg, ...)                                                             \
     LOG##level##MOD(blobmgr, "[traceID={},shardID=0x{:x},pg={},shard=0x{:x},blob={}] " msg, trace_id, shard_id,        \
-                    (shard_id >> homeobject::shard_width), (shard_id & homeobject::shard_mask), blob_id,               \
-                    ##__VA_ARGS__)
+                    (shard_id >> homeobject::shard_width), (shard_id & homeobject::shard_mask),                        \
+                    blob_id, ##__VA_ARGS__)
 
 #define BLOGT(trace_id, shard_id, blob_id, msg, ...) BLOG(TRACE, trace_id, shard_id, blob_id, msg, ##__VA_ARGS__)
 #define BLOGD(trace_id, shard_id, blob_id, msg, ...) BLOG(DEBUG, trace_id, shard_id, blob_id, msg, ##__VA_ARGS__)
@@ -650,6 +650,11 @@ void HSHomeObject::on_blob_del_commit(int64_t lsn, sisl::blob const& header, sis
         BLOGE(tid, msg_header->shard_id, *r_cast< blob_id_t const* >(key.cbytes()),
               "replication message header is corrupted with crc error, lsn={} header={}", lsn, msg_header->to_string());
         if (ctx) { ctx->promise_.setValue(folly::makeUnexpected(BlobError(BlobErrorCode::CHECKSUM_MISMATCH))); }
+
+        // TODO:: add RELEASE_ASSERT or proper error handling for corrupted replication message header
+        // RELEASE_ASSERT(false, "replication message header is corrupted with crc error, lsn={}, header={}",
+        // lsn,msg_header->to_string());
+
         return;
     }
 

src/lib/homestore_backend/hs_debug_inject.cpp

@@ -0,0 +1,160 @@
+#include "hs_debug_inject.hpp"
+
+#include <cstring>
+#include <random>
+
+#include <homestore/blkdata_service.hpp>
+#include <homestore/btree/btree_kv.hpp>
+#include <homestore/btree/btree_req.hpp>
+#include <homestore/homestore.hpp>
+
+#include "hs_homeobject.hpp"
+
+namespace homeobject {
+namespace hs_debug_inject {
+
+InjectResult delete_blob_from_index(HSHomeObject& ho, pg_id_t pg_id, shard_id_t shard_id, blob_id_t blob_id) {
+    auto hs_pg = ho.get_hs_pg(pg_id);
+    if (!hs_pg) { return folly::makeUnexpected(fmt::format("pg {} not found", pg_id)); }
+
+    BlobRouteKey blob_key{BlobRoute{shard_id, blob_id}};
+    BlobRouteValue out_value;
+    homestore::BtreeSingleRemoveRequest remove_req{&blob_key, &out_value};
+    auto status = hs_pg->index_table_->remove(remove_req);
+    if (status != homestore::btree_status_t::success) {
+        return folly::makeUnexpected(
+            fmt::format("failed to remove blob {} from index, status={}", blob_id, static_cast< int >(status)));
+    }
+    return folly::Unit();
+}
+
+InjectResult delete_shard_from_index(HSHomeObject& ho, pg_id_t pg_id, shard_id_t shard_id) {
+    auto hs_pg = ho.get_hs_pg(pg_id);
+    if (!hs_pg) { return folly::makeUnexpected(fmt::format("pg {} not found", pg_id)); }
+
+    auto start_key = BlobRouteKey{BlobRoute{shard_id, 0}};
+    auto end_key = BlobRouteKey{BlobRoute{shard_id, std::numeric_limits< uint64_t >::max()}};
+    homestore::BtreeRangeRemoveRequest< BlobRouteKey > range_remove_req{
+        homestore::BtreeKeyRange< BlobRouteKey >{std::move(start_key), true /* inclusive */, std::move(end_key),
+                                                 true /* inclusive */},
+        nullptr, std::numeric_limits< uint32_t >::max(),
+        [](homestore::BtreeKey const&, homestore::BtreeValue const&) -> bool { return true; }};
+
+    auto status = hs_pg->index_table_->remove(range_remove_req);
+    if (status != homestore::btree_status_t::success && status != homestore::btree_status_t::not_found) {
+        return folly::makeUnexpected(
+            fmt::format("failed to remove shard {} from index, status={}", shard_id, static_cast< int >(status)));
+    }
+    return folly::Unit();
+}
+
+InjectResult corrupt_blob_data(HSHomeObject& ho, pg_id_t pg_id, shard_id_t shard_id, blob_id_t blob_id) {
+    auto hs_pg = ho.get_hs_pg(pg_id);
+    if (!hs_pg) { return folly::makeUnexpected(fmt::format("pg {} not found", pg_id)); }
+
+    auto& data_service = homestore::data_service();
+    const auto blk_size = data_service.get_blk_size();
+
+    BlobRouteKey blob_key{BlobRoute{shard_id, blob_id}};
+    BlobRouteValue out_value;
+    homestore::BtreeSingleGetRequest blob_get_req{&blob_key, &out_value};
+    auto status = hs_pg->index_table_->get(blob_get_req);
+    if (status != homestore::btree_status_t::success) {
+        return folly::makeUnexpected(
+            fmt::format("blob {} not found in index, status={}", blob_id, static_cast< int >(status)));
+    }
+
+    auto pbas = out_value.pbas();
+    auto total_size = pbas.blk_count() * blk_size;
+    sisl::sg_list data_sgs;
+    data_sgs.size = total_size;
+    data_sgs.iovs.emplace_back(iovec{.iov_base = iomanager.iobuf_alloc(blk_size, total_size), .iov_len = total_size});
+
+    std::string err_msg;
+    try {
+        data_service.async_read(pbas, data_sgs, total_size)
+            .thenValue([&](auto&& err) {
+                if (err) { throw std::runtime_error(fmt::format("read failed: {}", err.message())); }
+
+                auto* data_ptr = reinterpret_cast< uint8_t* >(data_sgs.iovs[0].iov_base);
+                for (size_t i = 0; i < data_sgs.iovs[0].iov_len / 2; i++) {
+                    data_ptr[i] ^= 0xFF;
+                }
+
+                return data_service.async_write(data_sgs, pbas).thenValue([](auto&& werr) {
+                    if (werr) { throw std::runtime_error(fmt::format("write failed: {}", werr.message())); }
+                });
+            })
+            .get();
+    } catch (const std::exception& e) { err_msg = e.what(); }
+
+    iomanager.iobuf_free(reinterpret_cast< uint8_t* >(data_sgs.iovs[0].iov_base));
+    if (!err_msg.empty()) { return folly::makeUnexpected(err_msg); }
+    return folly::Unit();
+}
+
+InjectResult make_blob_inconsistent(HSHomeObject& ho, pg_id_t pg_id, shard_id_t shard_id, blob_id_t blob_id) {
+    auto hs_pg = ho.get_hs_pg(pg_id);
+    if (!hs_pg) { return folly::makeUnexpected(fmt::format("pg {} not found", pg_id)); }
+
+    auto& data_service = homestore::data_service();
+    const auto blk_size = data_service.get_blk_size();
+
+    BlobRouteKey blob_key{BlobRoute{shard_id, blob_id}};
+    BlobRouteValue out_value;
+    homestore::BtreeSingleGetRequest blob_get_req{&blob_key, &out_value};
+    auto status = hs_pg->index_table_->get(blob_get_req);
+    if (status != homestore::btree_status_t::success) {
+        return folly::makeUnexpected(
+            fmt::format("blob {} not found in index, status={}", blob_id, static_cast< int >(status)));
+    }
+
+    auto pbas = out_value.pbas();
+    auto total_size = pbas.blk_count() * blk_size;
+    sisl::sg_list data_sgs;
+    data_sgs.size = total_size;
+    data_sgs.iovs.emplace_back(iovec{.iov_base = iomanager.iobuf_alloc(blk_size, total_size), .iov_len = total_size});
+
+    std::string err_msg;
+    try {
+        data_service.async_read(pbas, data_sgs, total_size)
+            .thenValue([&](auto&& err) {
+                if (err) { throw std::runtime_error(fmt::format("read failed: {}", err.message())); }
+
+                uint8_t* read_buf = reinterpret_cast< uint8_t* >(data_sgs.iovs[0].iov_base);
+                auto* header = reinterpret_cast< HSHomeObject::BlobHeader* >(read_buf);
+                uint8_t* blob_bytes = read_buf + header->data_offset;
+
+                std::mt19937 rng{std::random_device{}()};
+                std::uniform_int_distribution< int > dist(0, 255);
+                for (size_t i = 0; i < header->blob_size / 2; i++) {
+                    blob_bytes[i] ^= static_cast< uint8_t >(dist(rng));
+                }
+
+                uint8_t computed_hash[HSHomeObject::BlobHeader::blob_max_hash_len]{};
+                ho.compute_blob_payload_hash(header->hash_algorithm, blob_bytes, header->blob_size, computed_hash,
+                                             HSHomeObject::BlobHeader::blob_max_hash_len);
+                std::memcpy(header->hash, computed_hash, HSHomeObject::BlobHeader::blob_max_hash_len);
+                std::memset(header->header_hash, 0, HSHomeObject::BlobHeader::blob_max_hash_len);
+                uint32_t computed_header_hash = crc32_ieee(0, (uint8_t*)header, sizeof(HSHomeObject::BlobHeader));
+                std::memcpy(header->header_hash, &computed_header_hash, sizeof(uint32_t));
+
+                if (!ho.verify_blob(data_sgs.iovs[0].iov_base, header->shard_id, header->blob_id)) {
+                    throw std::runtime_error(
+                        fmt::format("blob verification failed after modification, blob_id={}", blob_id));
+                }
+
+                return data_service.async_write(data_sgs, pbas).thenValue([](auto&& werr) {
+                    if (werr) { throw std::runtime_error(fmt::format("write failed: {}", werr.message())); }
+                });
+            })
+            .get();
+    } catch (const std::exception& e) { err_msg = e.what(); }
+
+    iomanager.iobuf_free(reinterpret_cast< uint8_t* >(data_sgs.iovs[0].iov_base));
+    if (!err_msg.empty()) { return folly::makeUnexpected(err_msg); }
+    return folly::Unit();
+}
+
+} // namespace hs_debug_inject
+} // namespace homeobject

src/lib/homestore_backend/hs_debug_inject.hpp

@@ -0,0 +1,35 @@
+#pragma once
+
+#include <folly/Expected.h>
+#include <folly/Unit.h>
+#include <string>
+
+#include "homeobject/common.hpp"
+
+namespace homeobject {
+
+class HSHomeObject;
+
+namespace hs_debug_inject {
+
+using InjectResult = folly::Expected< folly::Unit, std::string >;
+
+// Deletes a single blob's entry from its PG's local B-tree index, simulating a
+// missing blob on this replica without going through the replication layer.
+InjectResult delete_blob_from_index(HSHomeObject& ho, pg_id_t pg_id, shard_id_t shard_id, blob_id_t blob_id);
+
+// Deletes all blob entries for a shard from its PG's local B-tree index,
+// simulating a missing shard on this replica.
+InjectResult delete_shard_from_index(HSHomeObject& ho, pg_id_t pg_id, shard_id_t shard_id);
+
+// Flips bits in the first half of a blob's on-disk data, simulating silent
+// bit-rot. The blob's stored hash is left untouched, so scrub should report MISMATCH.
+InjectResult corrupt_blob_data(HSHomeObject& ho, pg_id_t pg_id, shard_id_t shard_id, blob_id_t blob_id);
+
+// Flips bits in a blob's on-disk data and recomputes a self-consistent hash,
+// simulating a replica whose copy differs from its peers but passes its own
+// integrity check.
+InjectResult make_blob_inconsistent(HSHomeObject& ho, pg_id_t pg_id, shard_id_t shard_id, blob_id_t blob_id);
+
+} // namespace hs_debug_inject
+} // namespace homeobject