chore: Enable TSAN without ignoring errors (#2828)

src/cluster/impl/RepeatedTask.hpp

@@ -30,7 +30,6 @@
 #include <boost/asio/spawn.hpp>
 #include <boost/asio/steady_timer.hpp>
 #include <boost/asio/strand.hpp>
-#include <boost/asio/use_future.hpp>
 
 #include <atomic>
 #include <chrono>
@@ -96,7 +95,12 @@ public:
         if (auto expected = State::Running; not state_.compare_exchange_strong(expected, State::Stopped))
             return;  // Already stopped or not started
 
-        boost::asio::spawn(strand_, [this](auto&&) { timer_.cancel(); }, boost::asio::use_future).wait();
+        std::binary_semaphore cancelSemaphore{0};
+        boost::asio::post(strand_, [this, &cancelSemaphore]() {
+            timer_.cancel();
+            cancelSemaphore.release();
+        });
+        cancelSemaphore.acquire();
         semaphore_.acquire();
     }
 };

src/feed/impl/SingleFeedBase.cpp

@@ -64,7 +64,7 @@ SingleFeedBase::unsub(SubscriberSharedPtr const& subscriber)
 void
 SingleFeedBase::pub(std::string msg)
 {
-    [[maybe_unused]] auto task = strand_.execute([this, msg = std::move(msg)]() {
+    strand_.submit([this, msg = std::move(msg)] {
         auto const msgPtr = std::make_shared<std::string>(msg);
         signal_.emit(msgPtr);
     });

src/feed/impl/TrackableSignal.hpp

@@ -73,10 +73,15 @@ public:
 
         // This class can't hold the trackable's shared_ptr, because disconnect should be able to be called in the
         // the trackable's destructor. However, the trackable can not be destroyed when the slot is being called
-        // either. track_foreign will hold a weak_ptr to the connection, which makes sure the connection is valid when
-        // the slot is called.
+        // either. `track_foreign` is racey when one shared_ptr is tracked by multiple signals. Therefore we are storing
+        // a weak_ptr of the trackable and using weak_ptr::lock() to atomically check existence and acquire a shared_ptr
+        // during slot invocation. This guarantees to keep the trackable alive for the duration of the slot call and
+        // avoids potential race conditions.
         connections->emplace(
-            trackable.get(), signal_.connect(typename SignalType::slot_type(slot).track_foreign(trackable))
+            trackable.get(), signal_.connect([slot, weakTrackable = std::weak_ptr(trackable)](Args&&... args) {
+                if (auto lifeExtender = weakTrackable.lock(); lifeExtender)
+                    std::invoke(slot, std::forward<Args...>(args)...);
+            })
         );
         return true;
     }