SDSTOR-21989: check index_table existence before using it (#409)

Author: JacksonYao287

conanfile.py

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

src/lib/homestore_backend/hs_blob_manager.cpp

@@ -88,7 +88,7 @@ BlobManager::AsyncResult< blob_id_t > HSHomeObject::_put_blob(ShardInfo const& s
         return folly::makeUnexpected(BlobErrorCode::SHUTTING_DOWN);
     }
     incr_pending_request_num();
-        // check user key size
+    // check user key size
     if (blob.user_key.size() > BlobHeader::max_user_key_length) {
         BLOGE(tid, shard.id, 0, "input user key length > max_user_key_length {}", blob.user_key.size(),
               BlobHeader::max_user_key_length);
@@ -167,8 +167,7 @@ BlobManager::AsyncResult< blob_id_t > HSHomeObject::_put_blob(ShardInfo const& s
 
     // Set offset of actual data after the blob header and user key (rounded off)
     req->blob_header()->data_offset = req->blob_header_buf().size();
-    RELEASE_ASSERT(req->blob_header()->data_offset == _data_block_size,
-                       "blob header should equals _data_block_size");
+    RELEASE_ASSERT(req->blob_header()->data_offset == _data_block_size, "blob header should equals _data_block_size");
     // In case blob body is not aligned, create a new aligned buffer and copy the blob body.
     if (((r_cast< uintptr_t >(blob.body.cbytes()) % io_align) != 0) || ((blob_size % io_align) != 0)) {
         // If address or size is not aligned, create a separate aligned buffer and do expensive memcpy.
@@ -367,9 +366,7 @@ BlobManager::AsyncResult< Blob > HSHomeObject::_get_blob_data(const shared< home
             }
 
             auto verify_result = do_verify_blob(read_buf.cbytes(), shard_id, 0 /* no blob_id check */);
-            if (!verify_result.hasValue()) {
-                return folly::makeUnexpected(verify_result.error());
-            }
+            if (!verify_result.hasValue()) { return folly::makeUnexpected(verify_result.error()); }
             std::string user_key = std::move(verify_result.value());
 
             BlobHeader const* header = r_cast< BlobHeader const* >(read_buf.cbytes());
@@ -498,8 +495,14 @@ HSHomeObject::blob_put_get_blk_alloc_hints(sisl::blob const& header, cintrusive<
           hs_shard->sb_->p_chunk_id, get_reserved_blks());
 
     if (msg_header->blob_id != 0) {
+        auto pg_index_table = hs_pg->index_table_;
+        if (!pg_index_table) {
+            LOGW("index table is not found for pg={}, skip statistics refresh", msg_header->pg_id);
+            return folly::makeUnexpected(homestore::ReplServiceError::RESULT_NOT_EXIST_YET);
+        }
+
         // check if the blob already exists, if yes, return the blk id
-        auto r = get_blob_from_index_table(hs_pg->index_table_, msg_header->shard_id, msg_header->blob_id);
+        auto r = get_blob_from_index_table(pg_index_table, msg_header->shard_id, msg_header->blob_id);
         if (r.hasValue()) {
             BLOGT(tid, msg_header->shard_id, msg_header->blob_id,
                   "Blob has already been persisted, blk_num={}, blk_count={}", r.value().blk_num(),

src/lib/homestore_backend/hs_pg_manager.cpp

@@ -1214,6 +1214,12 @@ uint32_t HSHomeObject::get_pg_tombstone_blob_count(pg_id_t pg_id) const {
         return 0;
     }
 
+    auto pg_index_table = hs_pg->index_table_;
+    if (!pg_index_table) {
+        LOGE("index table is not found for pg={}, skip statistics refresh", pg_id);
+        return 0;
+    }
+
     uint32_t tombstone_blob_count{0};
 
     auto start_key =
@@ -1234,7 +1240,7 @@ uint32_t HSHomeObject::get_pg_tombstone_blob_count(pg_id_t pg_id) const {
 
     std::vector< std::pair< BlobRouteKey, BlobRouteValue > > valid_blob_indexes;
 
-    auto const status = hs_pg->index_table_->query(query_req, valid_blob_indexes);
+    auto const status = pg_index_table->query(query_req, valid_blob_indexes);
     if (status != homestore::btree_status_t::success && status != homestore::btree_status_t::has_more) {
         LOGERROR("Failed to query blobs in index table for pg={}", pg_id);
         return 0;
@@ -1249,6 +1255,15 @@ uint32_t HSHomeObject::get_pg_tombstone_blob_count(pg_id_t pg_id) const {
 void HSHomeObject::refresh_pg_statistics(pg_id_t pg_id) {
     auto hs_pg = const_cast< HS_PG* >(_get_hs_pg_unlocked(pg_id));
     RELEASE_ASSERT(hs_pg, "Failed to get pg={} for statistics refresh", pg_id);
+    auto pg_index_table = hs_pg->index_table_;
+    if (!pg_index_table) {
+        if (hs_pg->pg_sb_->state == PGState::DESTROYED) {
+            LOGI("pg={} is destroyed, skip statistics refresh", pg_id);
+        } else {
+            RELEASE_ASSERT(false, "index table is not found for pg={} and not in PGState::DESTROYED state", pg_id);
+        }
+        return;
+    }
 
     // Step 1: Scan index table to count active and tombstone blobs in one pass
     uint64_t active_count = 0;
@@ -1276,7 +1291,7 @@ void HSHomeObject::refresh_pg_statistics(pg_id_t pg_id) {
         }};
 
     std::vector< std::pair< BlobRouteKey, BlobRouteValue > > dummy_out;
-    auto ret = hs_pg->index_table_->query(query_req, dummy_out);
+    auto ret = pg_index_table->query(query_req, dummy_out);
     RELEASE_ASSERT(ret == homestore::btree_status_t::success, "Failed to scan index table for pg={}, status={}", pg_id,
                    ret);