fix(hash): prevent new fields from inheriting stale key TTL after all fields expire

src/types/redis_hash.cc

@@ -463,6 +463,57 @@ rocksdb::Status Hash::scanAndRepair(engine::Context &ctx, const Slice &ns_key, H
   return rocksdb::Status::OK();
 }
 
+rocksdb::Status Hash::ensureValidMetadata(engine::Context &ctx, const std::string &ns_key,
+                                          HashMetadata *metadata, bool *was_reset) {
+  *was_reset = false;
+  if (!metadata->IsFieldExpirationEncoding()) return rocksdb::Status::OK();
+  if (metadata->persist > 0) return rocksdb::Status::OK();
+  // Defensive: ParseMetadata already returns NotFound for size==0 Hash,
+  // so this branch is unreachable in normal flow. Kept as a safety guard.
+  if (metadata->size == 0) return rocksdb::Status::OK();
+
+  uint64_t now = util::GetTimeStampMS();
+
+  // Fast path: if lower bound hasn't been reached, all TTL fields are still alive
+  if (metadata->lower != 0 && now < metadata->lower) {
+    return rocksdb::Status::OK();
+  }
+
+  // Fast path: if upper bound has passed, all TTL fields have expired
+  if (metadata->upper != 0 && now > metadata->upper) {
+    *metadata = createMetadataForWrite();
+    *was_reset = true;
+    return rocksdb::Status::OK();
+  }
+
+  // Slow path: scan for the first live field
+  std::string prefix_key = InternalKey(ns_key, "", metadata->version, storage_->IsSlotIdEncoded()).Encode();
+  std::string next_version_prefix_key =
+      InternalKey(ns_key, "", metadata->version + 1, storage_->IsSlotIdEncoded()).Encode();
+
+  rocksdb::ReadOptions read_options = ctx.DefaultScanOptions();
+  rocksdb::Slice lower_bound(prefix_key);
+  rocksdb::Slice upper_bound(next_version_prefix_key);
+  read_options.iterate_lower_bound = &lower_bound;
+  read_options.iterate_upper_bound = &upper_bound;
+
+  auto iter = util::UniqueIterator(ctx, read_options);
+  for (iter->Seek(prefix_key); iter->Valid() && iter->key().starts_with(prefix_key); iter->Next()) {
+    HashFieldState state;
+    auto s = DecodeFieldState(*metadata, Slice(iter->value()), now, &state);
+    if (!s.ok()) return s;
+    if (state.kind == HashFieldStateKind::kLiveTTL) {
+      return rocksdb::Status::OK();  
+    }
+  }
+  auto s = iter->status();
+  if (!s.ok()) return s;
+
+  *metadata = createMetadataForWrite();
+  *was_reset = true;
+  return rocksdb::Status::OK();
+}
+
 rocksdb::Status Hash::Get(engine::Context &ctx, const Slice &user_key, const Slice &field, std::string *value) {
   std::string ns_key = AppendNamespacePrefix(user_key);
   HashMetadata metadata(false);
@@ -494,6 +545,11 @@ rocksdb::Status Hash::IncrBy(engine::Context &ctx, const Slice &user_key, const
   HashMetadata metadata = createMetadataForWrite();
   rocksdb::Status s = getMetadata(ctx, ns_key, &metadata);
   if (!s.ok() && !s.IsNotFound()) return s;
+  if (s.ok()) {
+    bool was_reset = false;
+    s = ensureValidMetadata(ctx, ns_key, &metadata, &was_reset);
+    if (!s.ok()) return s;
+  }
 
   std::string sub_key = InternalKey(ns_key, field, metadata.version, storage_->IsSlotIdEncoded()).Encode();
   if (s.ok()) {
@@ -557,6 +613,11 @@ rocksdb::Status Hash::IncrByFloat(engine::Context &ctx, const Slice &user_key, c
   HashMetadata metadata = createMetadataForWrite();
   rocksdb::Status s = getMetadata(ctx, ns_key, &metadata);
   if (!s.ok() && !s.IsNotFound()) return s;
+  if (s.ok()) {
+    bool was_reset = false;
+    s = ensureValidMetadata(ctx, ns_key, &metadata, &was_reset);
+    if (!s.ok()) return s;
+  }
 
   std::string sub_key = InternalKey(ns_key, field, metadata.version, storage_->IsSlotIdEncoded()).Encode();
   if (s.ok()) {
@@ -664,6 +725,17 @@ rocksdb::Status Hash::SetFieldsWithExpire(engine::Context &ctx, const Slice &use
   HashMetadata metadata(false);
   auto s = getMetadata(ctx, ns_key, &metadata);
   bool metadata_existed = s.ok();
+  if (s.ok()) {
+    if (metadata.IsLegacySubkeyEncoding()) {
+      return rocksdb::Status::InvalidArgument(kHashFieldExpirationLegacyEncodingError);
+    }
+    bool was_reset = false;
+    s = ensureValidMetadata(ctx, ns_key, &metadata, &was_reset);
+    if (!s.ok()) return s;
+    if (was_reset) {
+      metadata_existed = false;
+    }
+  }
   if (s.IsNotFound()) {
     if (options.condition == HashFieldSetCondition::kFXX) return rocksdb::Status::OK();
     metadata = createMetadataForWrite();
@@ -672,8 +744,6 @@ rocksdb::Status Hash::SetFieldsWithExpire(engine::Context &ctx, const Slice &use
     }
   } else if (!s.ok()) {
     return s;
-  } else if (metadata.IsLegacySubkeyEncoding()) {
-    return rocksdb::Status::InvalidArgument(kHashFieldExpirationLegacyEncodingError);
   }
 
   s = ValidateHashFieldExpirationMetadata(metadata);
@@ -1024,6 +1094,11 @@ rocksdb::Status Hash::MSet(engine::Context &ctx, const Slice &user_key, const st
   HashMetadata metadata = createMetadataForWrite();
   rocksdb::Status s = getMetadata(ctx, ns_key, &metadata);
   if (!s.ok() && !s.IsNotFound()) return s;
+  if (s.ok()) {
+    bool was_reset = false;
+    s = ensureValidMetadata(ctx, ns_key, &metadata, &was_reset);
+    if (!s.ok()) return s;
+  }
   bool had_existing_fields = s.ok() && metadata.size > 0;
   bool ttl_updated = false;
   if (expire > 0 && metadata.expire != expire) {

src/types/redis_hash.h

@@ -129,6 +129,11 @@ class Hash : public SubKeyScanner {
   static rocksdb::Status decodeValue(const HashMetadata &metadata, Slice *value, uint64_t *expire = nullptr);
   rocksdb::Status scanAndRepair(engine::Context &ctx, const Slice &ns_key, HashMetadata *metadata, uint64_t now,
                                 uint64_t *size);
+  // Checks whether existing metadata is still valid when all fields may have expired via HFE.
+  // If metadata is stale (all fields expired), resets it to fresh metadata with a new version,
+  // clearing the stale key-level TTL. Sets *was_reset to true if metadata was reset.
+  rocksdb::Status ensureValidMetadata(engine::Context &ctx, const std::string &ns_key, HashMetadata *metadata,
+                                      bool *was_reset);
 
   friend struct FieldValueRetriever;
 };

tests/cppunit/types/hash_test.cc

@@ -1849,7 +1849,10 @@ TEST_F(RedisHashFieldExpirationEncodingTest, SetAndGetFieldsWithExpirePreserveEx
   expect_identity();
 }
 
-TEST_F(RedisHashFieldExpirationEncodingTest, SetFieldsWithExpireRetainsConservativeKeyTTLCornerState) {
+// After the fix for https://github.com/apache/kvrocks/issues/3635, when all fields have expired
+// (even if upper bound suggests otherwise), ensureValidMetadata resets metadata with a new version
+// and clears the stale key-level TTL.
+TEST_F(RedisHashFieldExpirationEncodingTest, SetFieldsWithExpireResetsMetadataWhenAllFieldsExpiredDespiteConservativeUpper) {
   const std::string key = "hsetex-key-ttl-conservative-upper";
   const uint64_t now = util::GetTimeStampMS();
   const uint64_t short_expire = now - 1'000;
@@ -1875,13 +1878,13 @@ TEST_F(RedisHashFieldExpirationEncodingTest, SetFieldsWithExpireRetainsConservat
   ASSERT_TRUE(s.ok()) << s.ToString();
   EXPECT_TRUE(applied);
   metadata = hashMetadata(key);
-  EXPECT_EQ(metadata.size, 2);
+  // ensureValidMetadata detected all fields expired and reset metadata
+  EXPECT_NE(metadata.version, original_metadata.version);
+  EXPECT_EQ(metadata.expire, 0);  // stale key-level TTL cleared
+  EXPECT_EQ(metadata.size, 1);
   EXPECT_EQ(metadata.persist, 1);
-  EXPECT_EQ(metadata.lower, short_expire);
-  EXPECT_EQ(metadata.upper, long_expire);
-  EXPECT_EQ(metadata.expire, original_metadata.expire);
-  EXPECT_EQ(metadata.version, original_metadata.version);
-  EXPECT_EQ(metadata.flags, original_metadata.flags);
+  EXPECT_EQ(metadata.lower, 0);
+  EXPECT_EQ(metadata.upper, 0);
   EXPECT_EQ(decodedHashValue(key, "new"), (std::pair<std::string, uint64_t>{"value", 0}));
 }
 
@@ -1940,6 +1943,113 @@ TEST_F(RedisHashFieldExpirationEncodingTest, SetFieldsWithExpireRecreatesExpired
   }
 }
 
+// Regression test for https://github.com/apache/kvrocks/issues/3635
+// When all fields expire via HFE, a subsequent HSET should NOT inherit the stale key-level TTL.
+TEST_F(RedisHashFieldExpirationEncodingTest, NewFieldDoesNotInheritStaleKeyTTLAfterAllFieldsExpire) {
+  const std::string key = "hfe-stale-key-ttl";
+  const uint64_t now = util::GetTimeStampMS();
+
+  // Step 1: Create hash with a persistent field
+  uint64_t ret = 0;
+  auto s = hash_->MSet(*ctx_, key, {{"old", "v"}}, false, &ret);
+  ASSERT_TRUE(s.ok()) << s.ToString();
+  ASSERT_EQ(ret, 1);
+
+  // Step 2: Set field-level TTL on "old" (120s)
+  std::vector<int64_t> results;
+  s = hash_->ExpireFields(*ctx_, key, {"old"}, now + 120'000, HashFieldExpireCondition::kNone, &results);
+  ASSERT_TRUE(s.ok()) << s.ToString();
+
+  // Step 3: Set key-level TTL (60s)
+  HashMetadata metadata = hashMetadata(key);
+  metadata.expire = now + 60'000;
+  s = putHashMetadata(key, metadata);
+  ASSERT_TRUE(s.ok()) << s.ToString();
+
+  // Step 4: Shorten field-level TTL to the past (force immediate expiry)
+  s = hash_->ExpireFields(*ctx_, key, {"old"}, now - 1, HashFieldExpireCondition::kNone, &results);
+  ASSERT_TRUE(s.ok()) << s.ToString();
+
+  // Step 5: Verify "old" field is expired (HGET returns NotFound)
+  std::string value;
+  s = hash_->Get(*ctx_, key, "old", &value);
+  EXPECT_TRUE(s.IsNotFound());
+
+  // Step 6: Write a new field via HSET
+  s = hash_->Set(*ctx_, key, "new", "v", &ret);
+  ASSERT_TRUE(s.ok()) << s.ToString();
+
+  // Step 7: The new field should NOT inherit the old key-level TTL
+  metadata = hashMetadata(key);
+  EXPECT_EQ(metadata.expire, 0) << "New field should not inherit stale key-level TTL";
+  EXPECT_EQ(metadata.size, 1);
+  EXPECT_EQ(metadata.persist, 1);
+
+  // Verify the new field is readable
+  s = hash_->Get(*ctx_, key, "new", &value);
+  ASSERT_TRUE(s.ok()) << s.ToString();
+  EXPECT_EQ(value, "v");
+}
+
+// Same scenario but via IncrBy
+TEST_F(RedisHashFieldExpirationEncodingTest, IncrByDoesNotInheritStaleKeyTTLAfterAllFieldsExpire) {
+  const std::string key = "hfe-stale-key-ttl-incr";
+  const uint64_t now = util::GetTimeStampMS();
+
+  uint64_t ret = 0;
+  auto s = hash_->MSet(*ctx_, key, {{"old", "10"}}, false, &ret);
+  ASSERT_TRUE(s.ok()) << s.ToString();
+
+  std::vector<int64_t> results;
+  s = hash_->ExpireFields(*ctx_, key, {"old"}, now - 1, HashFieldExpireCondition::kNone, &results);
+  ASSERT_TRUE(s.ok()) << s.ToString();
+
+  HashMetadata metadata = hashMetadata(key);
+  metadata.expire = now + 60'000;
+  s = putHashMetadata(key, metadata);
+  ASSERT_TRUE(s.ok()) << s.ToString();
+
+  // "old" is now expired
+  int64_t new_value = 0;
+  s = hash_->IncrBy(*ctx_, key, "old", 5, &new_value);
+  ASSERT_TRUE(s.ok()) << s.ToString();
+  EXPECT_EQ(new_value, 5);  // Treated as new field: 0 + 5
+
+  metadata = hashMetadata(key);
+  EXPECT_EQ(metadata.expire, 0) << "IncrBy should not inherit stale key-level TTL";
+}
+
+// Same scenario but via SetFieldsWithExpire
+TEST_F(RedisHashFieldExpirationEncodingTest, SetFieldsWithExpireDoesNotInheritStaleKeyTTLAfterAllFieldsExpire) {
+  const std::string key = "hfe-stale-key-ttl-hsetex";
+  const uint64_t now = util::GetTimeStampMS();
+
+  uint64_t ret = 0;
+  auto s = hash_->MSet(*ctx_, key, {{"old", "v"}}, false, &ret);
+  ASSERT_TRUE(s.ok()) << s.ToString();
+
+  std::vector<int64_t> results;
+  s = hash_->ExpireFields(*ctx_, key, {"old"}, now - 1, HashFieldExpireCondition::kNone, &results);
+  ASSERT_TRUE(s.ok()) << s.ToString();
+
+  HashMetadata metadata = hashMetadata(key);
+  metadata.expire = now + 60'000;
+  s = putHashMetadata(key, metadata);
+  ASSERT_TRUE(s.ok()) << s.ToString();
+
+  // "old" is now expired, write new field via SetFieldsWithExpire
+  bool applied = false;
+  auto options = setExOptions(HashSetExOptions::TTLAction::kDiscard);
+  s = hash_->SetFieldsWithExpire(*ctx_, key, {{"new", "v"}}, options, &applied, now);
+  ASSERT_TRUE(s.ok()) << s.ToString();
+  EXPECT_TRUE(applied);
+
+  metadata = hashMetadata(key);
+  EXPECT_EQ(metadata.expire, 0) << "SetFieldsWithExpire should not inherit stale key-level TTL";
+  EXPECT_EQ(metadata.size, 1);
+  EXPECT_EQ(metadata.persist, 1);
+}
+
 TEST_F(RedisHashTest, SetAndGetFieldsWithExpireRejectExistingLegacyHashWithoutMutation) {
   redis::Database db(storage_.get(), "hash_ns");
   const std::string key = "hsetex-legacy-existing";