Merge pull request ClickHouse#89566 from ClickHouse/backport/25.8/89496

Backport ClickHouse#89496 to 25.8: Fix changelog load in Keeper if rename failed

Author: antonio2368

src/Coordination/Changelog.cpp

@@ -1881,7 +1881,17 @@ try
             }
 
             ChangelogReader reader(changelog_description_ptr);
-            last_log_read_result = reader.readChangelog(entry_storage, start_to_read_from, log);
+            auto log_read_result = reader.readChangelog(entry_storage, start_to_read_from, log);
+
+            /// We didn't find the first required log in this changelog so we move to the next changelog
+            /// This can happen in case we failed to rename changelog to a name with correct first and last log index
+            if (log_read_result.first_read_index == 0)
+            {
+                LOG_TRACE(log, "Changelog contains only logs before {}", start_to_read_from);
+                continue;
+            }
+
+            last_log_read_result = std::move(log_read_result);
 
             if (last_log_read_result->last_read_index != 0)
                 last_read_index = last_log_read_result->last_read_index;

src/Coordination/tests/gtest_coordination_changelog.cpp

@@ -1149,7 +1149,6 @@ TYPED_TEST(CoordinationChangelogTest, TestRotateIntervalChanges)
 
 TYPED_TEST(CoordinationChangelogTest, ChangelogTestMaxLogSize)
 {
-
     ChangelogDirTest test("./logs");
     this->setLogDirectory("./logs");
 
@@ -1524,4 +1523,59 @@ TYPED_TEST(CoordinationChangelogTest, ChangelogTestBrokenWriteAt)
     }
 }
 
+TYPED_TEST(CoordinationChangelogTest, ChangelogLoadingFromInvalidName)
+{
+    if (this->enable_compression)
+        return;
+
+    ChangelogDirTest test("./logs");
+    this->setLogDirectory("./logs");
+
+    {
+        DB::KeeperLogStore changelog(
+            DB::LogFileSettings{
+                .force_sync = true, .compress_logs = this->enable_compression, .rotate_interval = 100'000, .max_size = 500},
+            DB::FlushSettings(),
+            this->keeper_context);
+        changelog.init(1, 0);
+
+        EXPECT_TRUE(fs::exists("./logs/changelog_1_100000.bin"));
+        for (size_t i = 0; i < 500; ++i)
+        {
+            auto entry = getLogEntry(std::to_string(i) + "_hello_world", 1);
+            changelog.append(entry);
+        }
+        changelog.end_of_append_batch(0, 0);
+
+        waitDurableLogs(changelog);
+    }
+
+    // Find file starting with "changelog_1_" (renamed because of file size limit)
+    fs::path new_changelog_path;
+    for (const auto & entry : fs::directory_iterator("./logs"))
+    {
+        if (entry.is_regular_file())
+        {
+            const auto filename = entry.path().filename().string();
+            if (filename.starts_with("changelog_1_"))
+                new_changelog_path = entry.path();
+        }
+    }
+
+    ASSERT_NE(new_changelog_path, fs::path{});
+
+    fs::rename(new_changelog_path, "./logs/changelog_1_100000.bin");
+
+    std::cout << new_changelog_path << std::endl;
+
+    DB::KeeperLogStore changelog(
+        DB::LogFileSettings{
+            .force_sync = true, .compress_logs = this->enable_compression, .rotate_interval = 100'000, .max_size = 500},
+        DB::FlushSettings(),
+        this->keeper_context);
+    changelog.init(15, 0);
+
+    ASSERT_EQ(changelog.next_slot(), 501);
+}
+
 #endif