This is an automated email from the ASF dual-hosted git repository.
git-hulk pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
The following commit(s) were added to refs/heads/unstable by this push:
new aa907d860 fix(replication): prevent full-sync checkpoint race with
cron purge (#3559)
aa907d860 is described below
commit aa907d860d549a0de60b01677c210a04ec7edcdd
Author: advisedy <[email protected]>
AuthorDate: Mon Jul 20 13:43:48 2026 +0800
fix(replication): prevent full-sync checkpoint race with cron purge (#3559)
Fixes #3100
## Problem
TSan reported an oversized allocation:
```
ThreadSanitizer: requested allocation size 0x100000000000000
exceeds maximum supported size of 0x10000000000
```
Root cause:
Before the fix, `GetFullReplDataInfo` released the lock before calling
`GetChildren`, while `Server::cron` called `rocksdb::DestroyDB` on the
live checkpoint directory without holding the lock at all. This allowed
the following interleaving:
```
meta: unlock checkpoint_mu_
purge: DestroyDB starts deleting files
meta: GetChildren → [] (dir exists but files are already gone)
```
`GetFullReplDataInfo` builds the file list by appending entries and then
calls `files->pop_back()`. If the directory is empty, `files` is an
empty string, and `pop_back` on an empty string is undefined behavior.
(https://en.cppreference.com/cpp/string/basic_string/pop_back)
On libstdc++, the common manifestation is a size underflow that produces
a very large number. The subsequent `files + CRLF` then allocates memory
based on this corrupted size, triggering the TSan error.
eaxmple:
```cpp
#include <cstdio>
#include <string>
int main() {
std::string files;
files.pop_back();
auto s = files + "\r\n";
printf("size=%zu\n", files.size());
return 0;
}
```
Running with GCC 11 + ASan on Linux:
```
AddressSanitizer: requested allocation size 0x100000000000000
(0x100000000001000 after adjustments for alignment, red zones etc.)
exceeds maximum supported size of 0x10000000000 (thread T0)
#0 0x7ffffee8b1e7 in operator new(unsigned long)
../../../../src/libsanitizer/asan/asan_new_delete.cpp:99
#1 0x7ffffecf8109 in void
std::__cxx11::basic_string::_M_construct<char*>(...)
(/lib/x86_64-linux-gnu/libstdc++.so.6+0x14f109)
...
SUMMARY: AddressSanitizer: allocation-size-too-big
```
The corrupted size (`0x100000000000000`) and the threshold
(`0x10000000000`) match the TSan report in the issue.
While investigating, we also found that the same race condition, besides
the "empty list" extreme case, could cause the replica to receive an
incomplete file list (`GetChildren` returned only a subset of files).
---
## Fix
No `pop_back` on empty list：
In `GetFullReplDataInfo`, if `files` is empty, return `NotOK`
immediately instead of calling `pop_back`.
Introduce `TryPurgeCheckpoint` — rename then destroy：
Added `Storage::TryPurgeCheckpoint`, which changes the purge logic to:
1. Under `checkpoint_mu_`, atomically rename `checkpoint_dir` to
`checkpoint_dir.trash` via `RenameFile`;
2. Clear `checkpoint_info_`;
3. After releasing the lock, call `DestroyDB(trash)`.
This way, the meta listing either sees a complete live directory or no
directory at all.
---
## Tests
Two new C++ gtests added (`storage_test.cc`):
- `GetFullReplDataInfoRejectsEmptyCheckpoint`: Creates a checkpoint →
clears the directory → asserts `NotOK`.
- `TryPurgeCheckpointPolicyAndAtomicRemove`: Verifies idle purge, rename
atomicity, and `checkpoint_info_` reset.
---------
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---
src/server/server.cc | 18 ++-------
src/storage/storage.cc | 49 +++++++++++++++++++++++
src/storage/storage.h | 2 +
tests/cppunit/storage_test.cc | 90 +++++++++++++++++++++++++++++++++++++++++++
4 files changed, 144 insertions(+), 15 deletions(-)
diff --git a/src/server/server.cc b/src/server/server.cc
index 78c578abd..48f6177d8 100644
--- a/src/server/server.cc
+++ b/src/server/server.cc
@@ -956,21 +956,9 @@ void Server::cron() {
// No replica uses this checkpoint, we can remove it.
if (counter != 0 && counter % 100 == 0) {
- int64_t create_time_secs = storage->GetCheckpointCreateTimeSecs();
- int64_t access_time_secs = storage->GetCheckpointAccessTimeSecs();
-
- if (storage->ExistCheckpoint()) {
- // TODO(shooterit): support to config the alive time of checkpoint
- int64_t now_secs = util::GetTimeStamp<std::chrono::seconds>();
- if ((GetFetchFileThreadNum() == 0 && now_secs - access_time_secs > 30)
||
- (now_secs - create_time_secs > 24 * 60 * 60)) {
- auto s = rocksdb::DestroyDB(config_->checkpoint_dir,
rocksdb::Options());
- if (!s.ok()) {
- WARN("[server] Fail to clean checkpoint, error: {}", s.ToString());
- } else {
- INFO("[server] Clean checkpoint successfully");
- }
- }
+ auto s = storage->TryPurgeCheckpoint(GetFetchFileThreadNum());
+ if (!s.IsOK()) {
+ WARN("[server] Fail to clean checkpoint, error: {}", s.Msg());
}
}
// check if DB need to be resumed every minute
diff --git a/src/storage/storage.cc b/src/storage/storage.cc
index 74a9bca3a..f328b04f3 100644
--- a/src/storage/storage.cc
+++ b/src/storage/storage.cc
@@ -1169,6 +1169,10 @@ Status
Storage::ReplDataManager::GetFullReplDataInfo(Storage *storage, std::stri
files->append(f);
files->push_back(',');
}
+ if (files->empty()) {
+ WARN("[storage] Checkpoint directory is empty");
+ return {Status::NotOK, "checkpoint directory is empty"};
+ }
files->pop_back();
return Status::OK();
@@ -1181,6 +1185,51 @@ bool Storage::ExistCheckpoint() {
bool Storage::ExistSyncCheckpoint() { return
env_->FileExists(config_->sync_checkpoint_dir).ok(); }
+Status Storage::TryPurgeCheckpoint(int fetch_file_threads) {
+ std::string trash_dir = config_->checkpoint_dir + ".trash";
+ if (env_->FileExists(trash_dir).ok()) {
+ auto s = rocksdb::DestroyDB(trash_dir, rocksdb::Options());
+ if (!s.ok()) {
+ WARN("[storage] Fail to clean stale checkpoint trash, error: {}",
s.ToString());
+ return {Status::NotOK, s.ToString()};
+ }
+ }
+
+ {
+ std::lock_guard<std::mutex> lg(checkpoint_mu_);
+ if (!env_->FileExists(config_->checkpoint_dir).ok()) {
+ return Status::OK();
+ }
+
+ // TODO(shooterit): support to config the alive time of checkpoint
+ int64_t create_time_secs = checkpoint_info_.create_time_secs;
+ int64_t access_time_secs = checkpoint_info_.access_time_secs;
+ int64_t now_secs = util::GetTimeStamp<std::chrono::seconds>();
+ bool should_purge =
+ (fetch_file_threads == 0 && now_secs - access_time_secs > 30) ||
(now_secs - create_time_secs > 24 * 60 * 60);
+ if (!should_purge) {
+ return Status::OK();
+ }
+
+ auto s = env_->RenameFile(config_->checkpoint_dir, trash_dir);
+ if (!s.ok()) {
+ WARN("[storage] Fail to rename checkpoint for purge, error: {}",
s.ToString());
+ return {Status::NotOK, s.ToString()};
+ }
+ checkpoint_info_.create_time_secs = 0;
+ checkpoint_info_.access_time_secs = 0;
+ checkpoint_info_.latest_seq = 0;
+ }
+
+ auto s = rocksdb::DestroyDB(trash_dir, rocksdb::Options());
+ if (!s.ok()) {
+ WARN("[storage] Fail to clean checkpoint, error: {}", s.ToString());
+ return {Status::NotOK, s.ToString()};
+ }
+ INFO("[storage] Clean checkpoint successfully");
+ return Status::OK();
+}
+
Status Storage::InWALBoundary(rocksdb::SequenceNumber seq) {
std::unique_ptr<rocksdb::TransactionLogIterator> iter;
auto s = GetWALIter(seq, &iter);
diff --git a/src/storage/storage.h b/src/storage/storage.h
index e199052b6..6778a12b1 100644
--- a/src/storage/storage.h
+++ b/src/storage/storage.h
@@ -352,6 +352,8 @@ class Storage {
bool ExistCheckpoint();
bool ExistSyncCheckpoint();
+ // Rename checkpoint to "*.trash" under checkpoint_mu_, then DestroyDB
outside the lock.
+ Status TryPurgeCheckpoint(int fetch_file_threads);
int64_t GetCheckpointCreateTimeSecs() const { return
checkpoint_info_.create_time_secs; }
void SetCheckpointAccessTimeSecs(int64_t t) {
checkpoint_info_.access_time_secs = t; }
int64_t GetCheckpointAccessTimeSecs() const { return
checkpoint_info_.access_time_secs; }
diff --git a/tests/cppunit/storage_test.cc b/tests/cppunit/storage_test.cc
index e6404e5d2..0c6e60ff5 100644
--- a/tests/cppunit/storage_test.cc
+++ b/tests/cppunit/storage_test.cc
@@ -170,3 +170,93 @@ TEST(Storage, ReplDataManagerRejectsUnsafeFilenames) {
std::filesystem::remove_all("test_repl_file_validation_dir", ec);
ASSERT_FALSE(ec);
}
+
+TEST(Storage, GetFullReplDataInfoRejectsEmptyCheckpoint) {
+ std::error_code ec;
+
+ const std::string test_dir = "test_empty_checkpoint";
+ Config config;
+ config.db_dir = test_dir + "/db";
+ config.checkpoint_dir = test_dir + "/checkpoint";
+ config.slot_id_encoded = false;
+
+ std::filesystem::remove_all(test_dir, ec);
+ ASSERT_FALSE(ec);
+ std::filesystem::create_directory(test_dir, ec);
+ ASSERT_FALSE(ec);
+
+ auto storage = std::make_unique<engine::Storage>(&config);
+ auto s = storage->Open();
+ ASSERT_TRUE(s.IsOK()) << s.Msg();
+
+ auto ctx = engine::Context(storage.get());
+ rocksdb::WriteBatch batch;
+ batch.Put("k", "v");
+ ASSERT_TRUE(storage->Write(ctx, rocksdb::WriteOptions(), &batch).ok());
+
+ std::string files;
+ s = engine::Storage::ReplDataManager::GetFullReplDataInfo(storage.get(),
&files);
+ ASSERT_TRUE(s.IsOK()) << s.Msg();
+
+ std::filesystem::remove_all(config.checkpoint_dir, ec);
+ ASSERT_FALSE(ec);
+ std::filesystem::create_directory(config.checkpoint_dir, ec);
+ ASSERT_FALSE(ec);
+
+ files.clear();
+ s = engine::Storage::ReplDataManager::GetFullReplDataInfo(storage.get(),
&files);
+ EXPECT_FALSE(s.IsOK());
+ EXPECT_TRUE(files.empty());
+
+ std::filesystem::remove_all(test_dir, ec);
+ ASSERT_FALSE(ec);
+}
+
+TEST(Storage, TryPurgeCheckpoint) {
+ std::error_code ec;
+
+ const std::string test_dir = "test_purge_checkpoint";
+ Config config;
+ config.db_dir = test_dir + "/db";
+ config.checkpoint_dir = test_dir + "/checkpoint";
+ config.slot_id_encoded = false;
+
+ std::filesystem::remove_all(test_dir, ec);
+ ASSERT_FALSE(ec);
+ std::filesystem::create_directory(test_dir, ec);
+ ASSERT_FALSE(ec);
+
+ auto storage = std::make_unique<engine::Storage>(&config);
+ auto s = storage->Open();
+ ASSERT_TRUE(s.IsOK()) << s.Msg();
+
+ auto ctx = engine::Context(storage.get());
+ rocksdb::WriteBatch batch;
+ batch.Put("k", "v");
+ ASSERT_TRUE(storage->Write(ctx, rocksdb::WriteOptions(), &batch).ok());
+
+ std::string files;
+ s = engine::Storage::ReplDataManager::GetFullReplDataInfo(storage.get(),
&files);
+ ASSERT_TRUE(s.IsOK()) << s.Msg();
+
+ storage->SetCheckpointAccessTimeSecs(storage->GetCheckpointAccessTimeSecs()
- 60);
+
+ s = storage->TryPurgeCheckpoint(/*fetch_file_threads=*/1);
+ ASSERT_TRUE(s.IsOK()) << s.Msg();
+ EXPECT_TRUE(storage->ExistCheckpoint());
+
+ s = storage->TryPurgeCheckpoint(/*fetch_file_threads=*/0);
+ ASSERT_TRUE(s.IsOK()) << s.Msg();
+ EXPECT_FALSE(storage->ExistCheckpoint());
+ EXPECT_FALSE(std::filesystem::exists(config.checkpoint_dir + ".trash"));
+ EXPECT_EQ(storage->GetCheckpointCreateTimeSecs(), 0);
+ EXPECT_EQ(storage->GetCheckpointAccessTimeSecs(), 0);
+
+ files.clear();
+ s = engine::Storage::ReplDataManager::GetFullReplDataInfo(storage.get(),
&files);
+ ASSERT_TRUE(s.IsOK()) << s.Msg();
+ EXPECT_TRUE(storage->ExistCheckpoint());
+
+ std::filesystem::remove_all(test_dir, ec);
+ ASSERT_FALSE(ec);
+}
\ No newline at end of file
