fix(bloom): crash due to pinnable slice in bloom filter

src/types/redis_bloom_chain.cc

@@ -195,27 +195,38 @@ rocksdb::Status BloomChain::InsertCommon(engine::Context &ctx, const Slice &user
     if (exist) {
       (*rets)[i] = BloomFilterAddResult::kExist;
     } else {
+      auto pinnable_slice_from_string = [](std::string data) -> rocksdb::PinnableSlice {
+        // This is a workaround for the issue that PinnableSlice does not support
+        // constructing from a temporary string.
+        rocksdb::PinnableSlice slice;
+        *slice.GetSelf() = std::move(data);
+        slice.PinSelf();
+        return slice;
+      };
+      auto strip_string_from_pinnable_slice = [](rocksdb::PinnableSlice &slice) -> std::string {
+        if (slice.GetSelf() != nullptr) {
+          return std::move(*slice.GetSelf());
+        }
+        return slice.ToString();
+      };
       if (metadata.size + 1 > metadata.GetCapacity()) {
         if (metadata.IsScaling()) {
           s = batch->Put(bf_key_list.back(), bf_data_list.back().ToStringView());
           if (!s.ok()) return s;
           std::string bf_data;
           s = createBloomFilterInBatch(ns_key, &metadata, batch, &bf_data);
           if (!s.ok()) return s;
-          rocksdb::PinnableSlice pin_slice;
-          *pin_slice.GetSelf() = std::move(bf_data);
-          pin_slice.PinSelf();
-          bf_data_list.push_back(std::move(pin_slice));
+          bf_data_list.push_back(pinnable_slice_from_string(std::move(bf_data)));
           bf_key_list.push_back(getBFKey(ns_key, metadata, metadata.n_filters - 1));
         } else {
           (*rets)[i] = BloomFilterAddResult::kFull;
           continue;
         }
       }
-      std::string data = bf_data_list.back().ToString();
+      auto &bf_data = bf_data_list.back();
+      std::string data = strip_string_from_pinnable_slice(bf_data);
       bloomAdd(item_hash_list[i], data);
-      *bf_data_list.back().GetSelf() = std::move(data);
-      bf_data_list.back().PinSelf();
+      bf_data = pinnable_slice_from_string(std::move(data));
       (*rets)[i] = BloomFilterAddResult::kOk;
       metadata.size += 1;
     }

tests/cppunit/types/bloom_chain_test.cc

@@ -27,10 +27,13 @@
 
 class RedisBloomChainTest : public TestBase {
  protected:
-  explicit RedisBloomChainTest() { sb_chain_ = std::make_unique<redis::BloomChain>(storage_.get(), "sb_chain_ns"); }
+  explicit RedisBloomChainTest() = default;
   ~RedisBloomChainTest() override = default;
 
-  void SetUp() override { key_ = "test_sb_chain_key"; }
+  void SetUp() override {
+    key_ = "test_sb_chain_key";
+    sb_chain_ = std::make_unique<redis::BloomChain>(storage_.get(), "sb_chain_ns");
+  }
   void TearDown() override {}
 
   std::unique_ptr<redis::BloomChain> sb_chain_;
@@ -48,8 +51,6 @@ TEST_F(RedisBloomChainTest, Reserve) {
   s = sb_chain_->Reserve(*ctx_, key_, capacity, error_rate, expansion);
   EXPECT_FALSE(s.ok());
   EXPECT_EQ(s.ToString(), "Invalid argument: the key already exists");
-
-  s = sb_chain_->Del(*ctx_, key_);
 }
 
 TEST_F(RedisBloomChainTest, BasicAddAndTest) {
@@ -79,5 +80,34 @@ TEST_F(RedisBloomChainTest, BasicAddAndTest) {
     EXPECT_TRUE(s.ok());
     EXPECT_EQ(exist, false);
   }
-  s = sb_chain_->Del(*ctx_, key_);
+}
+
+TEST_F(RedisBloomChainTest, DuplicateInsert) {
+  std::string insert_items[] = {"item1", "item2", "item3", "item101", "item202", "303"};
+  for (const auto& insert_item : insert_items) {
+    redis::BloomFilterAddResult ret = redis::BloomFilterAddResult::kOk;
+    auto s = sb_chain_->Add(*ctx_, key_, insert_item, &ret);
+    EXPECT_TRUE(s.ok());
+    EXPECT_EQ(ret, redis::BloomFilterAddResult::kOk);
+  }
+  std::vector<std::string> arrays;
+  for (int64_t idx = 0; idx < 1000; ++idx) {
+    arrays.push_back("itemx" + std::to_string(idx));
+  }
+  std::vector<redis::BloomFilterAddResult> results;
+  results.resize(arrays.size());
+  auto s = sb_chain_->MAdd(*ctx_, key_, arrays, &results);
+  EXPECT_TRUE(s.ok());
+  arrays.clear();
+  for (int64_t idx = 1000; idx < 2000; ++idx) {
+    arrays.push_back("itemx" + std::to_string(idx));
+  }
+  s = sb_chain_->MAdd(*ctx_, key_, arrays, &results);
+  EXPECT_TRUE(s.ok());
+  arrays.clear();
+  for (int64_t idx = 2000; idx < 3000; ++idx) {
+    arrays.push_back("itemx" + std::to_string(idx));
+  }
+  s = sb_chain_->MAdd(*ctx_, key_, arrays, &results);
+  EXPECT_TRUE(s.ok());
 }