fix: use master's wall time(physical time) as ttl in binlog

src/pika_binlog_transverter.cc

@@ -14,6 +14,7 @@
 #include "include/pika_command.h"
 #include "include/pika_define.h"
 #include "storage/storage.h"
+#include "storage/storage_time.h"
 
 
 uint32_t BinlogItem::exec_time() const { return exec_time_; }
@@ -84,6 +85,7 @@ bool PikaBinlogTransverter::BinlogDecode(BinlogType type, const std::string& bin
     LOG(ERROR) << "Binlog Item type error, expect type:" << type << " actualy type: " << binlog_type;
     return false;
   }
+  storage::g_storage_logictime->UpdateLogicTime(binlog_item->exec_time());
   pstd::GetFixed32(&binlog_str, &binlog_item->exec_time_);
   pstd::GetFixed32(&binlog_str, &binlog_item->term_id_);
   pstd::GetFixed64(&binlog_str, &binlog_item->logic_id_);

src/pika_client_conn.cc

@@ -164,6 +164,7 @@ std::shared_ptr<Cmd> PikaClientConn::DoCmd(const PikaCmdArgsType& argv, const st
       return c_ptr;
     }
     if (g_pika_server->readonly(current_db_) && opt != kCmdNameExec) {
+      storage::g_storage_logictime->SetProtectionMode(true);
       c_ptr->res().SetRes(CmdRes::kErrOther, "READONLY You can't write against a read only replica.");
       return c_ptr;
     }
@@ -185,6 +186,7 @@ std::shared_ptr<Cmd> PikaClientConn::DoCmd(const PikaCmdArgsType& argv, const st
       }
     }
   }
+  storage::g_storage_logictime->SetProtectionMode(false);
 
   // Process Command
   c_ptr->Execute();

src/storage/include/storage/storage_time.h

@@ -0,0 +1,44 @@
+//  Copyright (c) 2024-present The storage Authors.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same directory.
+
+#ifndef STORAGE_TIME_H
+#define STORAGE_TIME_H
+
+#include <atomic>
+#include <memory>
+#include <cstdint>
+#include <rocksdb/env.h>  // Assuming you're using RocksDB for time functions
+
+namespace storage {
+
+class LogicTime {
+ public:
+  // Constructor
+  LogicTime() : logic_time_(0), protection_mode_(false) {}
+
+  // Get the current logical time or the system time
+  int64_t Now();
+
+  // Update the logical time
+  void UpdateLogicTime(int64_t time);
+
+  // Get the protection mode status
+  bool ProtectionMode();
+
+  // Set the protection mode
+  void SetProtectionMode(bool on);
+
+ private:
+  std::atomic<int64_t> logic_time_; // Logical time value
+  std::atomic<bool> protection_mode_; // Protection mode flag
+};
+
+// Global unique pointer to LogicTime instance
+extern std::unique_ptr<LogicTime> g_storage_logictime;
+
+} // namespace storage
+
+#endif // STORAGE_TIME_H
+

src/storage/src/base_filter.h

@@ -27,9 +27,8 @@ class BaseMetaFilter : public rocksdb::CompactionFilter {
   BaseMetaFilter() = default;
   bool Filter(int level, const rocksdb::Slice& key, const rocksdb::Slice& value, std::string* new_value,
               bool* value_changed) const override {
-    int64_t unix_time;
-    rocksdb::Env::Default()->GetCurrentTime(&unix_time);
-    auto cur_time = static_cast<uint64_t>(unix_time);
+    int64_t unix_time = g_storage_logictime->Now();
+    auto cur_time = static_cast<uint32_t>(unix_time);
     /*
      * For the filtering of meta information, because the field designs of string
      * and list are different, their filtering policies are written separately.

src/storage/src/base_value_format.h

@@ -15,6 +15,7 @@
 #include "src/mutex.h"
 
 #include "pstd/include/env.h"
+#include "storage/storage_time.h"
 
 namespace storage {
 
@@ -134,8 +135,8 @@ class ParsedInternalValue {
     if (etime_ == 0) {
       return false;
     }
-    int64_t unix_time;
-    rocksdb::Env::Default()->GetCurrentTime(&unix_time);
+
+    int64_t unix_time = g_storage_logictime->Now();
     return etime_ < unix_time;
   }
 

src/storage/src/storage_time.cc

@@ -0,0 +1,37 @@
+//  Copyright (c) 2024-present, Qihoo, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same directory.
+
+#include <memory>
+#include "storage/storage_time.h"
+#include <glog/logging.h>
+
+namespace storage {
+
+std::unique_ptr<LogicTime> g_storage_logictime = std::make_unique<LogicTime>();
+
+int64_t LogicTime::Now() {
+  if (protection_mode_) {
+    return logic_time_;
+  }
+  int64_t t;
+  rocksdb::Env::Default()->GetCurrentTime(&t);
+  return t;
+}
+
+void LogicTime::UpdateLogicTime(int64_t time) {
+  logic_time_ = time;
+}
+
+bool LogicTime::ProtectionMode() {
+  return protection_mode_;
+}
+
+void LogicTime::SetProtectionMode(bool on) {
+  protection_mode_ = on;
+  LOG(INFO) << "Set protection mode to: " << on;
+}
+
+} // namespace storage
+