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/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();
 }
 

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,55 @@ 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;
+            round_requested = false;
+        }
+        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;
+        round_requested = false;
     }
     wake.notify_all();
     if (thread.joinable())
@@ -122,7 +167,7 @@ void CasGcScheduler::requestRoundSoon()
 {
     {
         std::lock_guard lock(mutex);
-        if (stopping || !thread.joinable())
+        if (scheduler_state != SchedulerState::Running)
             return;
         round_requested = true;
     }
@@ -299,9 +344,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 +378,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 +473,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>
@@ -162,13 +163,19 @@ 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.
     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
@@ -227,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<bool> loop_exited_on_terminal_for_test{false};
     std::atomic<bool> hb_exited_on_terminal_for_test{false};

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
@@ -439,6 +483,42 @@ TEST(CASGCStopStart, ConcurrentStopStartFromTwoThreadsStaysConsistent)
     storage->gcStop();
 }
 
+TEST(CASGCStopStart, RequestRoundSoonConcurrentWithStopStartDoesNotRace)
+{
+    auto backend = std::make_shared<InMemoryBackend>();
+    auto store = openPoolForTest(backend);
+    CasGcScheduler sched(store, std::chrono::seconds(3600), "CasGcRequestStopRaceTest", "ca-disk");
+    sched.start();
+
+    std::promise<void> 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