From 90bb796d6373e096be2a6065134b365e11574ec5 Mon Sep 17 00:00:00 2001
From: Antonio Andelic <antonio@clickhouse.com>
Date: Tue, 21 Nov 2023 12:08:04 +0000
Subject: [PATCH] Small Keeper fixes

---
 src/Coordination/Changelog.cpp             | 27 ++++++++++++++++------
 src/Coordination/KeeperSnapshotManager.cpp |  6 +++++
 src/Coordination/KeeperStateManager.cpp    | 10 +++++++-
 src/Coordination/KeeperStateManager.h      |  1 +
 4 files changed, 36 insertions(+), 8 deletions(-)

diff --git a/src/Coordination/Changelog.cpp b/src/Coordination/Changelog.cpp
index c28cc368ac0..f715a02db28 100644
--- a/src/Coordination/Changelog.cpp
+++ b/src/Coordination/Changelog.cpp
@@ -617,8 +617,13 @@ Changelog::Changelog(
 
     /// Load all files on changelog disks
 
+    std::unordered_set<DiskPtr> read_disks;
+
     const auto load_from_disk = [&](const auto & disk)
     {
+        if (read_disks.contains(disk))
+            return;
+
         LOG_TRACE(log, "Reading from disk {}", disk->getName());
         std::unordered_map<std::string, std::string> incomplete_files;
 
@@ -639,19 +644,25 @@ Changelog::Changelog(
         std::vector<std::string> changelog_files;
         for (auto it = disk->iterateDirectory(""); it->isValid(); it->next())
         {
-            if (it->name() == changelogs_detached_dir)
+            const auto & file_name = it->name();
+            if (file_name == changelogs_detached_dir)
                 continue;
 
-            if (it->name().starts_with(tmp_prefix))
+            if (file_name.starts_with(tmp_prefix))
             {
-                incomplete_files.emplace(it->name().substr(tmp_prefix.size()), it->path());
+                incomplete_files.emplace(file_name.substr(tmp_prefix.size()), it->path());
                 continue;
             }
 
-            if (clean_incomplete_file(it->path()))
-                continue;
-
-            changelog_files.push_back(it->path());
+            if (file_name.starts_with(DEFAULT_PREFIX))
+            {
+                if (!clean_incomplete_file(it->path()))
+                    changelog_files.push_back(it->path());
+            }
+            else
+            {
+                LOG_WARNING(log, "Unknown file found in log directory: {}", file_name);
+            }
         }
 
         for (const auto & changelog_file : changelog_files)
@@ -671,6 +682,8 @@ Changelog::Changelog(
 
         for (const auto & [name, path] : incomplete_files)
             disk->removeFile(path);
+
+        read_disks.insert(disk);
     };
 
     /// Load all files from old disks
diff --git a/src/Coordination/KeeperSnapshotManager.cpp b/src/Coordination/KeeperSnapshotManager.cpp
index a216272a9e1..98f490facf2 100644
--- a/src/Coordination/KeeperSnapshotManager.cpp
+++ b/src/Coordination/KeeperSnapshotManager.cpp
@@ -539,8 +539,12 @@ KeeperSnapshotManager::KeeperSnapshotManager(
     , storage_tick_time(storage_tick_time_)
     , keeper_context(keeper_context_)
 {
+    std::unordered_set<DiskPtr> read_disks;
     const auto load_snapshot_from_disk = [&](const auto & disk)
     {
+        if (read_disks.contains(disk))
+            return;
+
         LOG_TRACE(log, "Reading from disk {}", disk->getName());
         std::unordered_map<std::string, std::string> incomplete_files;
 
@@ -590,6 +594,8 @@ KeeperSnapshotManager::KeeperSnapshotManager(
 
         for (const auto & [name, path] : incomplete_files)
             disk->removeFile(path);
+
+        read_disks.insert(disk);
     };
 
     for (const auto & disk : keeper_context->getOldSnapshotDisks())
diff --git a/src/Coordination/KeeperStateManager.cpp b/src/Coordination/KeeperStateManager.cpp
index 879a42f6258..efe8a0cb2bd 100644
--- a/src/Coordination/KeeperStateManager.cpp
+++ b/src/Coordination/KeeperStateManager.cpp
@@ -269,6 +269,7 @@ KeeperStateManager::KeeperStateManager(
 void KeeperStateManager::loadLogStore(uint64_t last_commited_index, uint64_t logs_to_keep)
 {
     log_store->init(last_commited_index, logs_to_keep);
+    log_store_initialized = true;
 }
 
 void KeeperStateManager::system_exit(const int /* exit_code */)
@@ -361,6 +362,8 @@ void KeeperStateManager::save_state(const nuraft::srv_state & state)
 
 nuraft::ptr<nuraft::srv_state> KeeperStateManager::read_state()
 {
+    chassert(log_store_initialized);
+
     const auto & old_path = getOldServerStatePath();
 
     auto disk = getStateFileDisk();
@@ -454,7 +457,12 @@ nuraft::ptr<nuraft::srv_state> KeeperStateManager::read_state()
         disk->removeFile(copy_lock_file);
     }
 
-    LOG_WARNING(logger, "No state was read");
+    if (log_store->next_slot() != 1)
+        LOG_ERROR(
+            logger,
+            "No state was read but Keeper contains data which indicates that the state file was lost. This is dangerous and can lead to "
+            "data loss.");
+
     return nullptr;
 }
 
diff --git a/src/Coordination/KeeperStateManager.h b/src/Coordination/KeeperStateManager.h
index e402143c179..fd05261ac6c 100644
--- a/src/Coordination/KeeperStateManager.h
+++ b/src/Coordination/KeeperStateManager.h
@@ -121,6 +121,7 @@ private:
     mutable std::mutex configuration_wrapper_mutex;
     KeeperConfigurationWrapper configuration_wrapper TSA_GUARDED_BY(configuration_wrapper_mutex);
 
+    bool log_store_initialized = false;
     nuraft::ptr<KeeperLogStore> log_store;
 
     const String server_state_file_name;