more fixes, better test

Author: dranikpg

src/server/hset_family.cc

@@ -138,6 +138,14 @@ struct HMapWrap {
     VisitMut(ov);
   }
 
+  void Launder(tiering::SerializedMapDecoder* dec) {
+    Overloaded ov{
+        [](StringMap* s) {},
+        [&](detail::ListpackWrap& lw) { *dec->Write() = lw; },
+    };
+    VisitMut(ov);
+  }
+
   template <typename T> optional<T> Get() const {
     if (holds_alternative<T>(impl_))
       return get<T>(impl_);
@@ -244,9 +252,7 @@ template <typename F> auto ExecuteW(Transaction* tx, F&& f) {
         // Create wrapper from different types
         HMapWrap hw{*res.value()->Write()};
         fut.Resolve(f(hw));
-
-        // soak listpack wrapper back to get updated value
-        *res.value()->Write() = *hw.Get<detail::ListpackWrap>();
+        hw.Launder(*res);
       };
 
       es->tiered_storage()->Read(op_args.db_cntx.db_index, key, pv, D{}, std::move(read_cb));
@@ -330,6 +336,10 @@ OpStatus OpIncrBy(const OpArgs& op_args, string_view key, string_view field, Inc
 
   auto& add_res = *op_res;
   PrimeValue& pv = add_res.it->second;
+
+  if (pv.IsExternal() && !pv.IsCool())
+    return OpStatus::CANCELLED;  // Not supported for offloaded values
+
   if (add_res.is_new) {
     pv.InitRobj(OBJ_HASH, kEncodingListPack, lpNew(0));
   } else {
@@ -471,11 +481,12 @@ OpResult<CbVariant<uint32_t>> OpSet(const OpArgs& op_args, string_view key, CmdA
 
   // If the value is external, enqueue read and modify it there
   if (pv.IsExternal() && !pv.IsCool()) {
-    CHECK(op_sp.ttl == UINT32_MAX);  // TODO: remove
+    if (op_sp.ttl != UINT32_MAX)
+      return OpStatus::CANCELLED;  // Don't support expiry with offloaded hashes
+
     using D = tiering::SerializedMapDecoder;
     util::fb2::Future<OpResult<uint32_t>> fut;
     auto read_cb = [fut, values, &op_sp](io::Result<D*> res) mutable {
-      // Create wrapper from different types
       auto& lw = *res.value()->Write();
       uint32_t created = 0;
       for (size_t i = 0; i < values.size(); i += 2) {

src/server/tiered_storage_test.cc

@@ -493,8 +493,8 @@ TEST_F(PureDiskTSTest, Dump) {
 TEST_P(LatentCoolingTSTest, SimpleHash) {
   absl::FlagSaver saver;
   absl::SetFlag(&FLAGS_tiered_experimental_hash_support, true);
-  // For now, never upload as its not implemented yet
-  absl::SetFlag(&FLAGS_tiered_upload_threshold, 0.0);
+  absl::SetFlag(&FLAGS_tiered_upload_threshold,
+                0.0);  // For now, never upload as its not implemented yet
   UpdateFromFlags();
 
   const size_t kNUM = 100;
@@ -516,11 +516,9 @@ TEST_P(LatentCoolingTSTest, SimpleHash) {
   // Wait for all to be stashed or in end up in bins
   ExpectConditionWithinTimeout([=] {
     auto metrics = GetMetrics();
-    VLOG(0) << metrics.tiered_stats.total_stashes << " "
-            << metrics.tiered_stats.small_bins_entries_cnt;
-    return metrics.tiered_stats.total_stashes +
-               metrics.tiered_stats.small_bins_filling_entries_cnt ==
-           kNUM;
+    size_t sum =
+        metrics.tiered_stats.total_stashes + metrics.tiered_stats.small_bins_filling_entries_cnt;
+    return sum == kNUM;
   });
 
   // Verify correctness
@@ -533,20 +531,36 @@ TEST_P(LatentCoolingTSTest, SimpleHash) {
     EXPECT_EQ(resp, v);
   }
 
-  // Wait for all offloads again
-  ExpectConditionWithinTimeout([=] {
+  // Start offloading
+  SetFlag(&FLAGS_tiered_offload_threshold, 1.0);
+  UpdateFromFlags();
+  auto wait_offloaded = [=] {
     auto metrics = GetMetrics();
-    return metrics.db_stats[0].tiered_entries +
-               metrics.tiered_stats.small_bins_filling_entries_cnt ==
-           kNUM;
-  });
+    size_t sum =
+        metrics.db_stats[0].tiered_entries + metrics.tiered_stats.small_bins_filling_entries_cnt;
+    return sum == kNUM;
+  };
+
+  // Wait for all offloads again
+  ExpectConditionWithinTimeout(wait_offloaded);
 
   // HDEL
   for (size_t i = 0; i < kNUM; i++) {
     string key = absl::StrCat("k", i);
-    EXPECT_THAT(Run({"DEL", key, string{1, 'c'}}), IntArg(1));
+    EXPECT_THAT(Run({"HDEL", key, string{1, 'c'}}), IntArg(1));
     EXPECT_THAT(Run({"HLEN", key}), IntArg(25));
   }
+
+  // Wait for all offloads again
+  ExpectConditionWithinTimeout(wait_offloaded);
+
+  // HSET new field
+  for (size_t i = 0; i < kNUM; i++) {
+    string key = absl::StrCat("k", i);
+    EXPECT_THAT(Run({"HSET", key, string{1, 'c'}, "Some new value"}), IntArg(1));
+    EXPECT_THAT(Run({"HLEN", key}), IntArg(26));
+    EXPECT_EQ(Run({"HGET", key, string{1, 'c'}}), "Some new value");
+  }
 }
 
 }  // namespace dfly