From 74c6f96c49212c0b385acd952b66c45da5211511 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Tue, 8 Sep 2026 11:53:36 +0200 Subject: [PATCH 1/2] fix race Signed-off-by: Konstantin Morozov --- src/Common/FailPoint.cpp | 4 +- .../ContentAddressed/Gc/CasGcScheduler.cpp | 75 +++++++++++++++---- .../ContentAddressed/Gc/CasGcScheduler.h | 17 ++++- src/Disks/tests/gtest_cas_gc_stop_start.cpp | 44 +++++++++++ 4 files changed, 120 insertions(+), 20 deletions(-) diff --git a/src/Common/FailPoint.cpp b/src/Common/FailPoint.cpp index b05271d4f860..19a646648bdc 100644 --- a/src/Common/FailPoint.cpp +++ b/src/Common/FailPoint.cpp @@ -241,7 +241,9 @@ static struct InitFiu REGULAR(cas_relink_receiver_force_mechanism_failure) \ PAUSEABLE_ONCE(cas_relink_receiver_pause_before_confirm) \ REGULAR(cas_relink_sender_omit_pool_cookie) \ - REGULAR(cas_relink_receiver_drop_forced_disk) + REGULAR(cas_relink_receiver_drop_forced_disk) \ + ONCE(cas_gc_scheduler_fail_before_heartbeat_worker_start) \ + ONCE(cas_gc_scheduler_fail_before_worker_start) namespace FailPoints { diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp index 610d13879779..e0dee3413310 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp @@ -3,11 +3,13 @@ #include #include #include +#include #include #include #include #include #include +#include #include #include #include @@ -20,6 +22,13 @@ namespace DB::ErrorCodes extern const int TIMEOUT_EXCEEDED; extern const int SOCKET_TIMEOUT; extern const int MEMORY_LIMIT_EXCEEDED; + extern const int FAULT_INJECTED; +} + +namespace DB::FailPoints +{ + extern const char cas_gc_scheduler_fail_before_heartbeat_worker_start[]; + extern const char cas_gc_scheduler_fail_before_worker_start[]; } namespace DB::Cas @@ -90,19 +99,53 @@ CasGcScheduler::~CasGcScheduler() void CasGcScheduler::start() { - std::lock_guard lock(mutex); - if (thread.joinable()) - return; - stopping = false; - thread = ThreadFromGlobalPool([this] { loop(); }); - hb_thread = ThreadFromGlobalPool([this] { heartbeatLoop(); }); + std::lock_guard threads_lock(threads_mutex); + { + std::lock_guard lock(mutex); + if (scheduler_state == SchedulerState::Running) + return; + scheduler_state = SchedulerState::Running; + } + try + { + fiu_do_on(FailPoints::cas_gc_scheduler_fail_before_heartbeat_worker_start, + { + throw Exception(ErrorCodes::FAULT_INJECTED, "Injected failure before starting CAS GC heartbeat worker"); + }); + hb_thread = ThreadFromGlobalPool([this] { heartbeatLoop(); }); + fiu_do_on(FailPoints::cas_gc_scheduler_fail_before_worker_start, + { + throw Exception(ErrorCodes::FAULT_INJECTED, "Injected failure before starting CAS GC worker"); + }); + thread = ThreadFromGlobalPool([this] { loop(); }); + } + catch (...) + { + { + std::lock_guard lock(mutex); + scheduler_state = SchedulerState::Stopped; + } + wake.notify_all(); + if (thread.joinable()) + thread.join(); + if (hb_thread.joinable()) + hb_thread.join(); + i_am_leader.store(false, std::memory_order_relaxed); + throw; + } } void CasGcScheduler::stop() { + std::lock_guard threads_lock(threads_mutex); { std::lock_guard lock(mutex); - stopping = true; + if (scheduler_state == SchedulerState::Stopped) + { + i_am_leader.store(false, std::memory_order_relaxed); + return; + } + scheduler_state = SchedulerState::Stopped; } wake.notify_all(); if (thread.joinable()) @@ -122,7 +165,7 @@ void CasGcScheduler::requestRoundSoon() { { std::lock_guard lock(mutex); - if (stopping || !thread.joinable()) + if (scheduler_state != SchedulerState::Running) return; round_requested = true; } @@ -299,9 +342,10 @@ void CasGcScheduler::loop() while (true) { { - std::unique_lock lock(mutex); - wake.wait_for(lock, interval, [this] { return stopping || round_requested; }); - if (stopping) + UniqueLock lock(mutex); + wake.wait_for(lock.getUnderlyingLock(), interval, [this]() TSA_NO_THREAD_SAFETY_ANALYSIS + { return scheduler_state == SchedulerState::Stopped || round_requested; }); + if (scheduler_state == SchedulerState::Stopped) return; round_requested = false; } @@ -332,9 +376,9 @@ void CasGcScheduler::loop() } try { - /// LOW/benign: if stop() flips `stopping` while we're blocked here (a concurrent manual + /// LOW/benign: if stop() flips `scheduler_state` while we're blocked here (a concurrent manual /// round holds gc_round_mutex), we still run one more Scheduled round once it unblocks, - /// before the next wait_for() observes `stopping` - an accepted extra round, not a + /// before the next wait_for() observes `scheduler_state` - an accepted extra round, not a /// correctness issue. std::lock_guard round_lock(gc_round_mutex); @@ -427,8 +471,9 @@ void CasGcScheduler::heartbeatLoop() while (true) { { - std::unique_lock lock(mutex); - if (wake.wait_for(lock, hb_interval, [this] { return stopping; })) + UniqueLock lock(mutex); + if (wake.wait_for(lock.getUnderlyingLock(), hb_interval, [this]() TSA_NO_THREAD_SAFETY_ANALYSIS + { return scheduler_state == SchedulerState::Stopped; })) return; } /// rev.7 §3 [C1] + rev.8 §9 item 8: self-exit on ANY terminal (or FORGET-intent) pool, same as diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h index 80b50f2fc448..cebd691dd76a 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h @@ -3,6 +3,7 @@ #include #include #include +#include #include #include #include @@ -169,6 +170,12 @@ class CasGcScheduler bool waitForTerminalSelfExitForTest(std::chrono::milliseconds timeout); private: + enum class SchedulerState + { + Stopped, + Running + }; + /// Waits for the configured interval, runs scheduled rounds while the scheduler is active, and /// logs exceptions before continuing with the next tick. The round lock serializes this worker /// with `runOneRoundNow` because the persistent `gc` object is not thread-safe. @@ -210,15 +217,17 @@ class CasGcScheduler /// the round so stop()/heartbeatLoop are not blocked, so the round cannot hold `mutex`. std::mutex gc_round_mutex; + std::mutex threads_mutex; + ThreadFromGlobalPool thread TSA_GUARDED_BY(threads_mutex); + ThreadFromGlobalPool hb_thread TSA_GUARDED_BY(threads_mutex); + std::mutex mutex; std::condition_variable wake; - bool stopping = false; - bool round_requested = false; /// guarded by `mutex`; coalesced external wake request - ThreadFromGlobalPool thread; + SchedulerState scheduler_state TSA_GUARDED_BY(mutex) = SchedulerState::Stopped; + bool round_requested TSA_GUARDED_BY(mutex) = false; /// coalesced external wake request /// Set by the round worker and read by the heartbeat worker. It is only an in-process hint: the /// durable lease remains the authority, and a failed round clears the hint before retrying. std::atomic i_am_leader{false}; - ThreadFromGlobalPool hb_thread; /// Set true for the whole body of one round (`runRoundLogged`, held across the `gc_round_mutex` /// critical section a scheduled or manual round runs under) and cleared when it returns, on the diff --git a/src/Disks/tests/gtest_cas_gc_stop_start.cpp b/src/Disks/tests/gtest_cas_gc_stop_start.cpp index 6079a2b3436d..485862c84e96 100644 --- a/src/Disks/tests/gtest_cas_gc_stop_start.cpp +++ b/src/Disks/tests/gtest_cas_gc_stop_start.cpp @@ -8,6 +8,7 @@ #include #include #include +#include #include #include @@ -34,9 +35,16 @@ namespace DB::ErrorCodes { +extern const int FAULT_INJECTED; extern const int INVALID_STATE; } +namespace DB::FailPoints +{ +extern const char cas_gc_scheduler_fail_before_heartbeat_worker_start[]; +extern const char cas_gc_scheduler_fail_before_worker_start[]; +} + using namespace DB; using DB::Cas::CasGcScheduler; using DB::Cas::GcRoundLogRecord; @@ -342,6 +350,42 @@ TEST(CASGCStopStart, StopAndStartAreIdempotent) sched.stop(); } +TEST(CASGCStopStart, StopClearsLeadershipAfterManualRoundWithoutStart) +{ + auto backend = std::make_shared(); + auto store = openPoolForTest(backend); + CasGcScheduler sched(store, std::chrono::seconds(3600), "CasGcManualStopTest", "ca-disk"); + + const RoundReport report = sched.runOneRoundNow(); + ASSERT_TRUE(report.acquired_lease); + ASSERT_TRUE(sched.gcHealth().is_leader); + + sched.stop(); + EXPECT_FALSE(sched.gcHealth().is_leader); +} + +TEST(CASGCStopStart, StartFailureRollsBackAndCanBeRetried) +{ + for (const char * failpoint : + {FailPoints::cas_gc_scheduler_fail_before_heartbeat_worker_start, + FailPoints::cas_gc_scheduler_fail_before_worker_start}) + { + SCOPED_TRACE(failpoint); + auto backend = std::make_shared(); + auto store = openPoolForTest(backend); + CasGcScheduler sched(store, std::chrono::seconds(3600), "CasGcStartFailureTest", "ca-disk"); + + FailPointInjection::enableFailPoint(failpoint); + Cas::tests::expectThrowsCode(ErrorCodes::FAULT_INJECTED, [&] { sched.start(); }); + FailPointInjection::disableFailPoint(failpoint); + EXPECT_TRUE(sched.isQuiescent()); + + EXPECT_NO_THROW(sched.start()); + sched.stop(); + EXPECT_TRUE(sched.isQuiescent()); + } +} + /// (d) START refuses on a Vanished disk with the typed 668 (`INVALID_STATE`) error -- restarting GC on a /// decommissioned pool is meaningless and would only spin failing rounds -- while STOP on the SAME /// Vanished disk (with a live scheduler present) SUCCEEDS: stopping the reclaimer on a sick disk is a From e1d1b301b21557409e5a2cea4ba50b7b6c2eff23 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Wed, 16 Sep 2026 18:08:00 +0200 Subject: [PATCH 2/2] fix round requested Signed-off-by: Konstantin Morozov --- .../ContentAddressedMetadataStorage.cpp | 3 +- .../ContentAddressed/Gc/CasGcScheduler.cpp | 2 ++ .../ContentAddressed/Gc/CasGcScheduler.h | 4 +-- src/Disks/tests/gtest_cas_gc_stop_start.cpp | 36 +++++++++++++++++++ 4 files changed, 42 insertions(+), 3 deletions(-) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp index 78666e0ad297..8fb5a31b467c 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp @@ -1163,7 +1163,8 @@ void ContentAddressedMetadataStorage::gcStart() /// `start()` is a no-op if already running (idempotent) and re-enters the SAME instance after a stop -- /// the persistent `gc` observer + `gc_id` are preserved, and leadership is re-acquired only by the next /// round's normal `gc/state` acquisition, never restored here. Runs outside `pointer_mutex` for symmetry - /// with `stop()` (it spawns threads but joins nothing, so it does not block). + /// with `stop()`; it normally only spawns threads, but a worker-start failure joins any worker already + /// started during rollback and rethrows. snapshot->start(); } diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp index e0dee3413310..8b126b83d936 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp @@ -124,6 +124,7 @@ void CasGcScheduler::start() { std::lock_guard lock(mutex); scheduler_state = SchedulerState::Stopped; + round_requested = false; } wake.notify_all(); if (thread.joinable()) @@ -146,6 +147,7 @@ void CasGcScheduler::stop() return; } scheduler_state = SchedulerState::Stopped; + round_requested = false; } wake.notify_all(); if (thread.joinable()) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h index cebd691dd76a..ddab0711d6c2 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h @@ -163,7 +163,7 @@ class CasGcScheduler /// Test seam (rev.7 §3 [C1]): block up to `timeout` for BOTH the pacing and heartbeat loops to have /// SELF-EXITED via the terminal-lifecycle check — a `Vanished` pool or a published FORGET intent — as - /// opposed to exiting through `stop()`'s `stopping` flag. Returns false on timeout. Predicate-based + /// opposed to exiting through `stop()` transitioning `scheduler_state` to `Stopped`. Returns false on timeout. Predicate-based /// wait (no sleeps); the loops set their flag under `terminal_exit_mutex` before notifying, so there is /// no lost-wakeup window. Lets a test prove the self-exit path fired without relying on any wall-clock /// delay. @@ -236,7 +236,7 @@ class CasGcScheduler /// rev.7 §3 [C1] test-observation seam: set (under `terminal_exit_mutex`) by `loop`/`heartbeatLoop` /// respectively when they SELF-EXIT via the terminal-lifecycle check, NOT when `stop()` flips - /// `stopping`. `waitForTerminalSelfExitForTest` waits on `terminal_exit_cv` for BOTH, so a test proves + /// `scheduler_state` to `Stopped`. `waitForTerminalSelfExitForTest` waits on `terminal_exit_cv` for BOTH, so a test proves /// the self-exit path fired without any sleep. Purely diagnostic; production behavior never reads them. std::atomic loop_exited_on_terminal_for_test{false}; std::atomic hb_exited_on_terminal_for_test{false}; diff --git a/src/Disks/tests/gtest_cas_gc_stop_start.cpp b/src/Disks/tests/gtest_cas_gc_stop_start.cpp index 485862c84e96..1a529fb85375 100644 --- a/src/Disks/tests/gtest_cas_gc_stop_start.cpp +++ b/src/Disks/tests/gtest_cas_gc_stop_start.cpp @@ -483,6 +483,42 @@ TEST(CASGCStopStart, ConcurrentStopStartFromTwoThreadsStaysConsistent) storage->gcStop(); } +TEST(CASGCStopStart, RequestRoundSoonConcurrentWithStopStartDoesNotRace) +{ + auto backend = std::make_shared(); + auto store = openPoolForTest(backend); + CasGcScheduler sched(store, std::chrono::seconds(3600), "CasGcRequestStopRaceTest", "ca-disk"); + sched.start(); + + std::promise start; + const auto begin = start.get_future().share(); + + auto requester = std::async(std::launch::async, [&] + { + begin.wait(); + for (size_t i = 0; i < 1000; ++i) + sched.requestRoundSoon(); + }); + auto lifecycle = std::async(std::launch::async, [&] + { + begin.wait(); + for (size_t i = 0; i < 1000; ++i) + { + sched.stop(); + sched.start(); + } + }); + + start.set_value(); + ASSERT_EQ(requester.wait_for(std::chrono::seconds(60)), std::future_status::ready); + ASSERT_EQ(lifecycle.wait_for(std::chrono::seconds(60)), std::future_status::ready); + requester.get(); + lifecycle.get(); + + sched.stop(); + EXPECT_FALSE(sched.gcHealth().is_leader); +} + /// (T11 cannot-verify, acceptance matrix) Operator intent PERSISTS across a transient recovery: after the /// operator STOPs GC, the disk loses its mount lease (transient-not-live) and self-remounts back to Live — /// and NOTHING restarts the GC scheduler. Recovery is a Pool-internal operation with no reference to the