chore: Enable TSAN without ignoring errors (#2828)

Author: godexsoft

.github/scripts/execute-tests-under-sanitizer.sh

@@ -45,10 +45,6 @@ jobs:
 
     if: ${{ inputs.run_unit_tests }}
 
-    env:
-      # TODO: remove completely when we have fixed all currently existing issues with sanitizers
-      SANITIZER_IGNORE_ERRORS: ${{ endsWith(inputs.conan_profile, '.tsan') }}
-
     steps:
       - name: Cleanup workspace
         if: ${{ runner.os == 'macOS' }}
@@ -65,45 +61,27 @@ jobs:
       - name: Make clio_tests executable
         run: chmod +x ./clio_tests
 
-      - name: Run clio_tests (regular)
-        if: ${{ env.SANITIZER_IGNORE_ERRORS == 'false' }}
+      - name: Run clio_tests
+        continue-on-error: true
+        id: run_clio_tests
         run: ./clio_tests
 
-      - name: Run clio_tests (sanitizer errors ignored)
-        if: ${{ env.SANITIZER_IGNORE_ERRORS == 'true' }}
-        run: ./.github/scripts/execute-tests-under-sanitizer.sh ./clio_tests
-
-      - name: Check for sanitizer report
-        if: ${{ env.SANITIZER_IGNORE_ERRORS == 'true' }}
-        id: check_report
-        run: |
-          if ls .sanitizer-report/* 1> /dev/null 2>&1; then
-            echo "found_report=true" >> $GITHUB_OUTPUT
-          else
-            echo "found_report=false" >> $GITHUB_OUTPUT
-          fi
-
-      - name: Upload sanitizer report
-        if: ${{ env.SANITIZER_IGNORE_ERRORS == 'true' && steps.check_report.outputs.found_report == 'true' }}
-        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6.0.0
-        with:
-          name: sanitizer_report_${{ runner.os }}_${{ inputs.build_type }}_${{ inputs.conan_profile }}
-          path: .sanitizer-report/*
-          include-hidden-files: true
-
       - name: Create an issue
-        if: ${{ false && env.SANITIZER_IGNORE_ERRORS == 'true' && steps.check_report.outputs.found_report == 'true' }}
+        if: ${{ steps.run_clio_tests.outcome == 'failure' && endsWith(inputs.conan_profile, 'san') }}
         uses: ./.github/actions/create-issue
         env:
           GH_TOKEN: ${{ github.token }}
         with:
           labels: "bug"
           title: "[${{ inputs.conan_profile }}] reported issues"
           body: >
-            Clio tests failed one or more sanitizer checks when built with ${{ inputs.conan_profile }}`.
+            Clio tests failed one or more sanitizer checks when built with `${{ inputs.conan_profile }}`.
 
             Workflow: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}/
-            Reports are available as artifacts.
+
+      - name: Fail the job if clio_tests failed
+        if: ${{ steps.run_clio_tests.outcome == 'failure' }}
+        run: exit 1
 
   integration_tests:
     name: Integration testing

.github/workflows/reusable-test.yml

@@ -15,7 +15,6 @@ on:
       - ".github/actions/**"
       - "!.github/actions/build-docker-image/**"
       - "!.github/actions/create-issue/**"
-      - .github/scripts/execute-tests-under-sanitizer.sh
 
       - CMakeLists.txt
       - conanfile.py

.github/workflows/sanitizers.yml

@@ -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 @@ class RepeatedTask {
         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/cluster/impl/RepeatedTask.hpp

@@ -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/SingleFeedBase.cpp

@@ -73,10 +73,15 @@ class TrackableSignal {
 
         // 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;
     }

src/feed/impl/TrackableSignal.hpp

@@ -72,11 +72,11 @@ class SubscriptionManagerBaseTest : public util::prometheus::WithPrometheus, pub
     }
 
     StrictMockAmendmentCenterSharedPtr mockAmendmentCenterPtr_;
-    std::shared_ptr<SubscriptionManager> subscriptionManagerPtr_ =
-        std::make_shared<SubscriptionManager>(Execution(2), backend_, mockAmendmentCenterPtr_);
     web::SubscriptionContextPtr session_ = std::make_shared<MockSession>();
     MockSession* sessionPtr_ = dynamic_cast<MockSession*>(session_.get());
     uint32_t const networkID_ = 123;
+    std::shared_ptr<SubscriptionManager> subscriptionManagerPtr_ =
+        std::make_shared<SubscriptionManager>(Execution(2), backend_, mockAmendmentCenterPtr_);
 };
 
 using SubscriptionManagerTest = SubscriptionManagerBaseTest<util::async::SyncExecutionContext>;

tests/unit/feed/SubscriptionManagerTests.cpp

@@ -209,7 +209,7 @@ TYPED_TEST(ExecutionContextTests, repeatingOperation)
 {
     auto const repeatDelay = std::chrono::milliseconds{1};
     auto const timeout = std::chrono::milliseconds{15};
-    auto callCount = 0uz;
+    std::atomic_size_t callCount = 0uz;
 
     auto res = this->ctx.executeRepeatedly(repeatDelay, [&] { ++callCount; });
     auto timeSpent = util::timed([timeout] { std::this_thread::sleep_for(timeout); });  // calculate actual time spent