CAS: gc scheduler stop join race

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
 {

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp

@@ -3,11 +3,13 @@
 #include <Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Primitives/CasTypes.h>
 #include <Common/CurrentThread.h>
 #include <Common/Exception.h>
+#include <Common/FailPoint.h>
 #include <Common/ProfileEvents.h>
 #include <Common/ProfileEventsScope.h>
 #include <Common/logger_useful.h>
 #include <Common/setThreadName.h>
 #include <Common/thread_local_rng.h>
+#include <Common/UniqueLock.h>
 #include <base/scope_guard.h>
 #include <algorithm>
 #include <optional>
@@ -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

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h

@@ -3,6 +3,7 @@
 #include <Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h>
 #include <Common/ThreadPool.h>
 #include <base/types.h>
+#include <base/defines.h>
 #include <atomic>
 #include <chrono>
 #include <condition_variable>
@@ -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<bool> 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

src/Disks/tests/gtest_cas_gc_stop_start.cpp

@@ -8,6 +8,7 @@
 #include <Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h>
 #include <Disks/tests/cas_test_helpers.h>
 #include <Common/Exception.h>
+#include <Common/FailPoint.h>
 
 #include <atomic>
 #include <chrono>
@@ -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<InMemoryBackend>();
+    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<InMemoryBackend>();
+        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