Make delete and put api idempotent in recovery case.

Fixes in UTs to make with work with homestore.
If put fails and returns existing valid pba, then ignore.
If key not found for delete, then ignore. If key exists
and already tomebestone, then ignore. If key exists and
valid pba then do the free blks.

conanfile.py

@@ -9,7 +9,7 @@
 
 class HomeObjectConan(ConanFile):
     name = "homeobject"
-    version = "1.0.1"
+    version = "1.0.2"
     homepage = "https://github.com/eBay/HomeObject"
     description = "Blob Store built on HomeReplication"
     topics = ("ebay")

src/lib/homestore_backend/hs_blob_manager.cpp

@@ -363,14 +363,22 @@ void HSHomeObject::on_blob_del_commit(int64_t lsn, sisl::blob const& header, sis
 
     auto r = move_to_tombstone(index_table, blob_info);
     if (!r) {
-        LOGW("fail to move blob to tombstone,  blob_id {}, shard_id {}, {}", blob_info.blob_id, blob_info.shard_id,
-             r.error());
-        if (ctx) ctx->promise_.setValue(folly::makeUnexpected(r.error()));
-        return;
+        if (recovery_done_) {
+            LOGE("fail to move blob to tombstone,  blob_id {}, shard_id {}, {}", blob_info.blob_id, blob_info.shard_id,
+                 r.error());
+            if (ctx) ctx->promise_.setValue(folly::makeUnexpected(r.error()));
+            return;
+        } else {
+            if (ctx) { ctx->promise_.setValue(BlobManager::Result< BlobInfo >(blob_info)); }
+            return;
+        }
     }
 
     auto& multiBlks = r.value();
-    repl_dev->async_free_blks(lsn, multiBlks);
+    if (multiBlks != tombstone_pbas) {
+        repl_dev->async_free_blks(lsn, multiBlks);
+    }
+
     if (ctx) { ctx->promise_.setValue(BlobManager::Result< BlobInfo >(blob_info)); }
 }
 

src/lib/homestore_backend/hs_homeobject.cpp

@@ -1,4 +1,3 @@
-
 #include <latch>
 #include <optional>
 #include <spdlog/fmt/bin_to_hex.h>
@@ -142,6 +141,7 @@ void HSHomeObject::init_homestore() {
         RELEASE_ASSERT(new_id == _our_id, "Received new SvcId [{}] AFTER recovery of [{}]?!", to_string(new_id),
                        to_string(_our_id));
     }
+    recovery_done_ = true;
     LOGI("Initialize and start HomeStore is successfully");
 }
 

src/lib/homestore_backend/hs_homeobject.hpp

@@ -186,6 +186,7 @@ class HSHomeObject : public HomeObjectImpl {
 private:
     shared< HeapChunkSelector > chunk_selector_;
     iomgr::timer_handle_t ho_timer_thread_handle_;
+    bool recovery_done_{false};
 
 private:
     static homestore::ReplicationService& hs_repl_service() { return homestore::hs()->repl_service(); }

src/lib/homestore_backend/index_kv.cpp

@@ -43,10 +43,18 @@ HSHomeObject::recover_index_table(homestore::superblk< homestore::index_table_sb
 BlobManager::NullResult HSHomeObject::add_to_index_table(shared< BlobIndexTable > index_table,
                                                          const BlobInfo& blob_info) {
     BlobRouteKey index_key{BlobRoute{blob_info.shard_id, blob_info.blob_id}};
-    BlobRouteValue index_value{blob_info.pbas};
-    homestore::BtreeSinglePutRequest put_req{&index_key, &index_value, homestore::btree_put_type::INSERT};
+    BlobRouteValue index_value{blob_info.pbas}, existing_value;
+    homestore::BtreeSinglePutRequest put_req{&index_key, &index_value, homestore::btree_put_type::INSERT,
+                                             &existing_value};
     auto status = index_table->put(put_req);
-    if (status != homestore::btree_status_t::success) { return folly::makeUnexpected(BlobError::INDEX_ERROR); }
+    if (status != homestore::btree_status_t::success) {
+        if (existing_value.pbas().is_valid() || existing_value.pbas() == tombstone_pbas) {
+            // Check if the blob id already exists in the index or its tombstone.
+            return folly::Unit();
+        }
+        LOGE("Failed to put to index table error {}", status);
+        return folly::makeUnexpected(BlobError::INDEX_ERROR);
+    }
 
     return folly::Unit();
 }
@@ -75,7 +83,7 @@ BlobManager::Result< homestore::MultiBlkId > HSHomeObject::move_to_tombstone(sha
     homestore::BtreeSingleGetRequest get_req{&index_key, &index_value_get};
     auto status = index_table->get(get_req);
     if (status != homestore::btree_status_t::success) {
-        LOGDEBUG("Failed to get from index table [route={}]", index_key);
+        LOGE("Failed to get from index table [route={}]", index_key);
         return folly::makeUnexpected(BlobError::UNKNOWN_BLOB);
     }
 

src/lib/homestore_backend/tests/hs_blob_tests.cpp

@@ -46,6 +46,8 @@ class HomeObjectFixture : public ::testing::Test {
         _obj_inst = homeobject::init_homeobject(std::weak_ptr< homeobject::HomeObjectApplication >(app));
     }
 
+    void TearDown() override { app->clean(); }
+
     void create_pg(pg_id_t pg_id) {
         auto info = homeobject::PGInfo(pg_id);
         auto peer1 = _obj_inst->our_uuid();
@@ -194,6 +196,14 @@ TEST_F(HomeObjectFixture, BasicPutGetDelBlobWRestart) {
         LOGINFO("delete blob shard {} blob {}", shard_id, blob_id);
     }
 
+    // Delete again should have no errors.
+    for (const auto& [id, blob] : blob_map) {
+        int64_t shard_id = std::get< 1 >(id), blob_id = std::get< 2 >(id);
+        auto g = _obj_inst->blob_manager()->del(shard_id, blob_id).get();
+        ASSERT_TRUE(g);
+        LOGINFO("delete blob shard {} blob {}", shard_id, blob_id);
+    }
+
     // After delete all blobs, get should fail
     for (const auto& [id, blob] : blob_map) {
         int64_t shard_id = std::get< 1 >(id), blob_id = std::get< 2 >(id);

src/lib/homestore_backend/tests/hs_shard_tests.cpp

@@ -140,6 +140,8 @@ class ShardManagerTestingRecovery : public ::testing::Test {
 public:
     void SetUp() override { app = std::make_shared< FixtureAppWithRecovery >(); }
 
+    void TearDown() override { app->clean(); }
+
 protected:
     std::shared_ptr< FixtureApp > app;
 };

src/lib/tests/fixture_app.cpp

@@ -69,6 +69,8 @@ void TestFixture::SetUp() {
     });
 }
 
+void TestFixture::TearDown() { std::dynamic_pointer_cast< FixtureApp >(app)->clean(); }
+
 int main(int argc, char* argv[]) {
     int parsed_argc = argc;
     ::testing::InitGoogleTest(&parsed_argc, argv);

src/lib/tests/fixture_app.hpp

@@ -23,7 +23,7 @@ class FixtureApp : public homeobject::HomeObjectApplication {
 
 public:
     FixtureApp();
-    ~FixtureApp() override { clean(); }
+    ~FixtureApp() = default;
 
     bool spdk_mode() const override { return false; }
     uint32_t threads() const override { return 2; }
@@ -54,6 +54,7 @@ class TestFixture : public ::testing::Test {
     blob_id_t _blob_id;
 
     void SetUp() override;
+    void TearDown() override;
 
 protected:
     std::shared_ptr< homeobject::HomeObject > homeobj_;