cleanWal by commitLogId (vesoft-inc#4015)

src/kvstore/Listener.cpp

@@ -123,6 +123,11 @@ bool Listener::preProcessLog(LogID logId,
   return true;
 }
 
+void Listener::cleanWal() {
+  std::lock_guard<std::mutex> g(raftLock_);
+  wal()->cleanWAL(lastApplyLogId_);
+}
+
 std::tuple<nebula::cpp2::ErrorCode, LogID, TermID> Listener::commitLogs(
     std::unique_ptr<LogIterator> iter, bool) {
   LogID lastId = kNoCommitLogId;

src/kvstore/Listener.h

@@ -138,6 +138,11 @@ class Listener : public raftex::RaftPart {
     return lastApplyLogId_;
   }
 
+  /**
+   * @brief clean wal that before lastApplyLogId_
+   */
+  void cleanWal() override;
+
   /**
    * @brief clean up data in listener, called in RaftPart::reset
    *

src/kvstore/NebulaStore.cpp

@@ -1195,16 +1195,16 @@ void NebulaStore::cleanWAL() {
       auto& part = partEntry.second;
       if (part->needToCleanWal()) {
         // clean wal by expired time
-        part->wal()->cleanWAL();
+        part->cleanWal();
       }
     }
   }
   for (const auto& spaceEntry : spaceListeners_) {
     for (const auto& partEntry : spaceEntry.second->listeners_) {
       for (const auto& typeEntry : partEntry.second) {
         const auto& listener = typeEntry.second;
-        // clean wal by log id
-        listener->wal()->cleanWAL(listener->getApplyId());
+        // clean wal by commit log id
+        listener->cleanWal();
       }
     }
   }

src/kvstore/raftex/RaftPart.cpp

@@ -350,6 +350,11 @@ void RaftPart::stop() {
   VLOG(1) << idStr_ << "Partition has been stopped";
 }
 
+void RaftPart::cleanWal() {
+  std::lock_guard<std::mutex> g(raftLock_);
+  wal()->cleanWAL(committedLogId_);
+}
+
 nebula::cpp2::ErrorCode RaftPart::canAppendLogs() {
   DCHECK(!raftLock_.try_lock());
   if (UNLIKELY(status_ != Status::RUNNING)) {

src/kvstore/raftex/RaftPart.h

@@ -160,6 +160,11 @@ class RaftPart : public std::enable_shared_from_this<RaftPart> {
     return term_;
   }
 
+  /**
+   * @brief clean wal that before commitLogId
+   */
+  virtual void cleanWal();
+
   /**
    * @brief Return the wal
    */

src/kvstore/wal/FileBasedWal.cpp

@@ -672,23 +672,30 @@ void FileBasedWal::cleanWAL() {
 
 void FileBasedWal::cleanWAL(LogID id) {
   std::lock_guard<std::mutex> g(walFilesMutex_);
+  auto now = time::WallClock::fastNowInSec();
+  size_t index = 0;
   if (walFiles_.empty()) {
     return;
   }
+  auto size = walFiles_.size();
+  if (size < 2) {
+    return;
+  }
 
   if (walFiles_.rbegin()->second->lastId() < id) {
     VLOG(3) << "Try to clean wal not existed " << id << ", lastWal is "
             << walFiles_.rbegin()->second->lastId();
     return;
   }
-
+  int walTTL = FLAGS_wal_ttl;
   // remove wal file that lastId is less than target
   auto iter = walFiles_.begin();
   while (iter != walFiles_.end()) {
-    if (iter->second->lastId() < id) {
+    if (iter->second->lastId() < id && index < size - 2 && (now - iter->second->mtime() > walTTL)) {
       VLOG(3) << "Clean wals, Remove " << iter->second->path();
       unlink(iter->second->path());
       iter = walFiles_.erase(iter);
+      index++;
     } else {
       break;
     }

src/kvstore/wal/test/FileBasedWalTest.cpp

@@ -558,7 +558,7 @@ TEST(FileBasedWal, CleanWalBeforeIdTest) {
     CHECK_LE(wal->firstLogId(), logToKeep);
     CHECK(!wal->walFiles_.empty());
     CHECK_LE(wal->walFiles_.begin()->second->firstId(), logToKeep);
-    CHECK_GE(wal->walFiles_.begin()->second->lastId(), logToKeep);
+    CHECK_GE(wal->walFiles_.rbegin()->second->lastId(), logToKeep);
     logToKeep += folly::Random::rand64(10);
   }
   CHECK_EQ(1000, wal->lastLogId());