-sfix(rdb_load): replica load expired keys

src/server/rdb_load.cc

@@ -2129,6 +2129,7 @@ error_code RdbLoader::LoadKeyValPair(int type, ObjSettings* settings) {
   if (item == nullptr) {
     item = new Item;
   }
+  auto cleanup = absl::Cleanup([item] { delete item; });
 
   // Read key
   SET_OR_RETURN(ReadKey(), item->key);
@@ -2148,22 +2149,23 @@ error_code RdbLoader::LoadKeyValPair(int type, ObjSettings* settings) {
    * Similarly if the RDB is the preamble of an AOF file, we want to
    * load all the keys as they are, since the log of operations later
    * assume to work in an exact keyspace state. */
-  // TODO: check rdbflags&RDBFLAGS_AOF_PREAMBLE logic in rdb.c
-  bool should_expire = settings->has_expired;  // TODO: to implement
-  if (should_expire) {
-    // decrRefCount(val);
-  } else {
-    ShardId sid = Shard(item->key, shard_set->size());
-    item->expire_ms = settings->expiretime;
 
-    auto& out_buf = shard_buf_[sid];
+  if (ServerState::tlocal()->is_master && settings->has_expired) {
+    VLOG(1) << "Expire key: " << item->key;
+    return kOk;
+  }
 
-    out_buf.emplace_back(item);
+  ShardId sid = Shard(item->key, shard_set->size());
+  item->expire_ms = settings->expiretime;
 
-    constexpr size_t kBufSize = 128;
+  auto& out_buf = shard_buf_[sid];
-    if (out_buf.size() >= kBufSize) {
+
-      FlushShardAsync(sid);
+  out_buf.emplace_back(item);
-    }
+  std::move(cleanup).Cancel();
+
+  constexpr size_t kBufSize = 128;
+  if (out_buf.size() >= kBufSize) {
+    FlushShardAsync(sid);
   }
 
   return kOk;

src/server/rdb_save.h

@@ -186,7 +186,6 @@ class RdbSerializer {
   std::unique_ptr<LZF_HSLOT[]> lzf_;
 
   CompressionMode compression_mode_;
-  // TODO : This compressor impl should support different compression algorithms zstd/lz4 etc.
   std::unique_ptr<CompressorImpl> compressor_impl_;
 
   static constexpr size_t kMinStrSizeToCompress = 256;

src/server/rdb_test.cc

@@ -231,6 +231,15 @@ TEST_F(RdbTest, ReloadTtl) {
   EXPECT_LT(990, CheckedInt({"ttl", "key"}));
 }
 
+TEST_F(RdbTest, ReloadExpired) {
+  Run({"set", "key", "val"});
+  Run({"expire", "key", "2"});
+  sleep(2);
+  Run({"debug", "reload"});
+  auto resp = Run({"get", "key"});
+  ASSERT_THAT(resp, ArgType(RespExpr::NIL));
+}
+
 TEST_F(RdbTest, SaveFlush) {
   Run({"debug", "populate", "500000"});
 

src/server/replica.cc

@@ -43,8 +43,6 @@ using absl::StrCat;
 
 namespace {
 
-// TODO: 2. Use time-out on socket-reads so that we would not deadlock on unresponsive master.
-//       3. Support ipv6 at some point.
 int ResolveDns(std::string_view host, char* dest) {
   struct addrinfo hints, *servinfo;