Fix hangs in CI builds

.github/workflows/build_dependencies.yml

@@ -228,17 +228,31 @@ jobs:
         key_prefix: HomestoreDeps-${{ inputs.platform }}-${{ inputs.build-type }}-${{ inputs.malloc-impl }}
       if: ${{ github.event_name != 'pull_request' && steps.restore-cache.outputs.cache-hit != 'true' && inputs.tooling != 'Coverage'}}
 
+    - name: Install gdb
+      # src/tests/stack_dump_launcher.sh uses it to print the stacks of a hung test, and of a crashed one from the
+      # core.<pid> file it leaves in the test directory
+      run: |
+        command -v gdb || (sudo apt-get update -q && sudo apt-get install -y -q gdb)
+        sudo sysctl -w kernel.core_pattern=core.%p
+      if: ${{ inputs.testing == 'True' }}
+
     - name: Create and Test Package
       run: |
         if [[ "${{ inputs.tooling }}" == "AddressSanitize" ]]; then sanitize="address"
         elif [[ "${{ inputs.tooling }}" == "ThreadSanitize" ]]; then sanitize="thread"
         else sanitize="False"; fi
-        conan create \
+        # ASan makes the templated btree tests much bigger to compile: four parallel compiles have run the runner out of
+        # memory, killing the job mid-build.
+        build_jobs=""
+        if [[ "${sanitize}" == "address" ]]; then build_jobs="-c tools.build:jobs=2"; fi
+        conan create ${build_jobs} \
             -o sisl/*:malloc_impl=${{ inputs.malloc-impl }} \
             -o homestore/*:sanitize=${sanitize} \
             -s:h build_type=${{ inputs.build-type }} \
             ${{ inputs.conan-channel != '' && format('--user oss --channel {0}', inputs.conan-channel) || '' }} \
             .
+      env:
+        CTEST_OUTPUT_ON_FAILURE: 1
       if: ${{ inputs.testing == 'True' && inputs.tooling != 'Coverage' }}
 
     - name: Code Coverage Run
@@ -251,6 +265,8 @@ jobs:
             ${{ inputs.conan-channel != '' && format('--user oss --channel {0}', inputs.conan-channel) || '' }} \
             .
         gcovr --cobertura ./coverage.xml
+      env:
+        CTEST_OUTPUT_ON_FAILURE: 1
       if: ${{ inputs.testing == 'True' && inputs.tooling == 'Coverage' }}
 
     - name: Upload coverage reports to Codecov
@@ -259,3 +275,22 @@ jobs:
         files: ./coverage.xml
         disable_search: true
       if: ${{ inputs.tooling == 'Coverage' }}
+
+    - name: Collect Test Logs
+      # Also after a failed, timed out or cancelled test step: the test logs are the only record of what went wrong.
+      # Pack them first: the log directories are named by time (logs/2026-10-20_19:45), and upload-artifact rejects ':'.
+      run: |
+        shopt -s nullglob
+        dirs=(~/.conan2/p/b/homes*/b/build/*/Testing/Temporary ~/.conan2/p/b/homes*/b/build/*/src/tests/logs
+              build/*/Testing/Temporary build/*/src/tests/logs)
+        if (( ${#dirs[@]} )); then tar -czf test-logs.tgz --ignore-failed-read "${dirs[@]}" || true; fi
+      if: ${{ always() && inputs.testing == 'True' }}
+
+    - name: Upload Test Logs
+      uses: actions/upload-artifact@v4
+      with:
+        name: test-logs-${{ inputs.platform }}-${{ inputs.build-type }}-${{ inputs.malloc-impl }}-${{ inputs.tooling }}
+        path: test-logs.tgz
+        if-no-files-found: ignore
+        retention-days: 7
+      if: ${{ always() && inputs.testing == 'True' }}

CMakeLists.txt

@@ -32,15 +32,13 @@ if ($ENV{ASAN_OPTIONS} MATCHES "detect_leaks=1")
   set(ASAN_ENABLED_BUILD true)
 endif()
 
-if (DEFINED MEMORY_SANITIZER_ON)
-    if (${MEMORY_SANITIZER_ON})
-        message(STATUS "Build with Memory Sanitizer ON")
-        include (cmake/mem_sanitizer.cmake)
-    else ()
-        message(STATUS "Build with Memory Sanitizer OFF")
-    endif()
+# conanfile.py sets ADDRESS_SANITIZER_ON from the sanitize option; MEMORY_SANITIZER_ON is the older name for it.
+# THREAD_SANITIZER_ON isn't supported yet: with uninstrumented dependencies TSan reports thousands of false races.
+if (ADDRESS_SANITIZER_ON OR MEMORY_SANITIZER_ON)
+    message(STATUS "Build with Address Sanitizer ON")
+    include (cmake/mem_sanitizer.cmake)
 else ()
-    message(STATUS "Build with Memory Sanitizer OFF")
+    message(STATUS "Build with Address Sanitizer OFF")
 endif ()
 
 if (${CMAKE_BUILD_TYPE} STREQUAL "Debug")

conanfile.py

@@ -9,7 +9,7 @@
 
 class HomestoreConan(ConanFile):
     name = "homestore"
-    version = "8.3.5"
+    version = "8.3.6"
 
     homepage = "https://github.com/eBay/Homestore"
     description = "HomeStore Storage Engine"

src/include/homestore/logstore/log_store.hpp

@@ -302,6 +302,8 @@ class home_log_store : public std::enable_shared_from_this< home_log_store > {
     LogStoreServiceMetrics& m_metrics;
 
     logdev_key m_trunc_ld_key{0, 0};
+    // Serializes truncate(); see the lock order note there.
+    std::mutex m_truncate_mtx;
 
 private:
     // graceful shutdown related fields

src/include/homestore/replication/repl_dev.hpp

@@ -662,10 +662,16 @@ class repl_dev {
     // we have no shutdown for repl_dev, since shutdown repl_dev is done by repl_service
     void stop() {
         start_stopping();
-        while (true) {
+        for (uint64_t waited_sec{0};; ++waited_sec) {
             auto pending_request_num = get_pending_request_num();
             if (!pending_request_num) break;
 
+            // No deadline: callers free resources (e.g. the raft server in leave_group()) that in-flight requests
+            // still use. Report instead, so a leaked count shows up in the logs rather than as a silent hang.
+            if (waited_sec && (waited_sec % 30 == 0)) {
+                LOGWARNMOD(replication, "repl_dev group_id={} still waiting for {} pending requests after {} sec",
+                           boost::uuids::to_string(group_id()), pending_request_num, waited_sec);
+            }
             std::this_thread::sleep_for(std::chrono::milliseconds(1000));
         }
     }

src/lib/blkdata_svc/blk_read_tracker.cpp

@@ -52,11 +52,12 @@ void BlkReadTracker::merge(const blk_id& blkid, int64_t new_ref_count,
             // trying to acquire the same write lock in its IO completion path.
             boost::container::small_vector< blk_track_waiter_ptr, 8 > fired_waiters;
             m_pending_reads_map.upsert_or_delete(
-                base_blkid, [new_ref_count, &base_blkid, &fired_waiters](BlkTrackRecord& rec, bool existing) {
+                base_blkid, [this, new_ref_count, &base_blkid, &fired_waiters](BlkTrackRecord& rec, bool existing) {
                     HS_DBG_ASSERT_EQ(existing, true, "Decrement a ref count (blk: {}) which does not exist in map",
                                      base_blkid.to_string());
                     rec.m_ref_cnt += new_ref_count;
                     if (rec.m_ref_cnt == 0) {
+                        std::lock_guard lg{m_waiters_mtx};
                         fired_waiters = std::move(rec.m_waiters);
                         return true;
                     }
@@ -65,7 +66,8 @@ void BlkReadTracker::merge(const blk_id& blkid, int64_t new_ref_count,
             // fired_waiters destructs here, outside the write lock -> m_cb() fires safely
         } else {
             // this is wait_on operation
-            m_pending_reads_map.update(base_blkid, [&waiter_rescheduled, &waiter](BlkTrackRecord& rec) {
+            m_pending_reads_map.update(base_blkid, [this, &waiter_rescheduled, &waiter](BlkTrackRecord& rec) {
+                std::lock_guard lg{m_waiters_mtx};
                 rec.m_waiters.push_back(waiter);
                 waiter_rescheduled = true;
             });

src/lib/blkdata_svc/blk_read_tracker.hpp

@@ -14,6 +14,7 @@
  *********************************************************************************/
 #pragma once
 #include <functional>
+#include <mutex>
 
 #include <boost/container/small_vector.hpp>
 #include <sisl/cache/simple_hashmap.hpp>
@@ -115,6 +116,9 @@ class BlkReadTracker {
 
 private:
     sisl::SimpleHashMap< blk_id, BlkTrackRecord > m_pending_reads_map;
+    // wait_on() appends to BlkTrackRecord::m_waiters through SimpleHashMap::update(), which only holds the bucket's
+    // shared lock, so concurrent wait_on()s on the same record need this to modify m_waiters.
+    std::mutex m_waiters_mtx;
     BlkReadTrackerMetrics m_metrics;
     uint32_t m_entries_per_record{s_entries_per_record};
 

src/lib/checkpoint/cp_internal.hpp

@@ -36,7 +36,9 @@ class CPWatchdog {
     CP* m_cp;
     CPManager* m_cp_mgr;
     std::shared_mutex m_cp_mtx;
-    Clock::time_point m_last_state_ch_time;
+    Clock::time_point m_last_progress_time; // start of the current stall, reported if the cp is declared stuck
+    Clock::time_point m_last_repair_time;   // last repair_slow_cp() round, gives the repairs time to take effect
+    uint32_t m_repair_attempts{0};          // repair_slow_cp() rounds in the current stall
     uint64_t m_timer_sec{0};
     iomgr::timer_handle_t m_timer_hdl;
     uint32_t m_progress_pct{0};

src/lib/checkpoint/cp_mgr.cpp

@@ -448,7 +448,9 @@ void CPWatchdog::reset_cp() {
 void CPWatchdog::set_cp(CP* cp) {
     std::unique_lock< std::shared_mutex > lk{m_cp_mtx};
     m_cp = cp;
-    m_last_state_ch_time = Clock::now();
+    m_last_progress_time = Clock::now();
+    m_last_repair_time = m_last_progress_time;
+    m_repair_attempts = 0;
 }
 
 void CPWatchdog::stop() {
@@ -466,45 +468,52 @@ void CPWatchdog::cp_watchdog_timer() {
     // check if any cp to track
     if (m_cp == nullptr) { return; }
     const auto status = m_cp->get_status();
-    if ((status != cp_status_t::cp_flush_prepare) || (status != cp_status_t::cp_flushing)) { return; }
+    if ((status != cp_status_t::cp_flush_prepare) && (status != cp_status_t::cp_flushing)) { return; }
 
-    uint32_t cum_pct{0};
-    uint32_t count{0};
+    // Track the slowest consumer, so one wedged consumer can't be masked by its peers advancing.
+    // NOTE: every consumer currently reports a constant 100 (they flush synchronously), so in practice this is a
+    // time-based check: a cp that stays in flush for 12 timer periods is declared stuck.
+    uint32_t pct{100};
     for (auto& consumer : m_cp_mgr->consumer_list()) {
-        if (consumer) {
-            ++count;
-            cum_pct += consumer->cp_progress_percent();
-        }
+        if (consumer) { pct = std::min(pct, static_cast< uint32_t >(consumer->cp_progress_percent())); }
     }
-    if (m_progress_pct > cum_pct / count) {
+    if (pct > m_progress_pct) {
         // We are making progress in flushing the data.
-        m_progress_pct = cum_pct / count;
+        m_progress_pct = pct;
+        m_last_progress_time = Clock::now();
+        m_repair_attempts = 0;
         return;
     }
 
-    if (get_elapsed_time_ms(m_last_state_ch_time) >= m_timer_sec * 1000) {
-        LOGINFO("cp progress percent {} is not changed. time elapsed {}, cp state={} ", m_progress_pct,
-                get_elapsed_time_ms(m_last_state_ch_time), m_cp->to_string());
-    }
-
-    // check if enough time passed since last state change
+    // check if enough time passed since the last progress, and since the last repair attempt
+    const auto elapsed_ms = get_elapsed_time_ms(m_last_progress_time);
     uint32_t max_time_multiplier = 12;
-    if (get_elapsed_time_ms(m_last_state_ch_time) < max_time_multiplier * m_timer_sec * 1000) {
+    const auto max_stall_ms = max_time_multiplier * m_timer_sec * 1000;
+    if ((elapsed_ms < max_stall_ms) || (get_elapsed_time_ms(m_last_repair_time) < max_stall_ms)) { return; }
+
+    // Ask the consumers that are behind to push harder, but only a limited number of times: a consumer that is
+    // deadlocked never reaches 100, and repairing it forever would leave the cp stuck with just warn logs.
+    const uint32_t max_repair_attempts = 3;
+    if (m_repair_attempts < max_repair_attempts) {
         uint32_t repair_attempted{0};
         for (auto& consumer : m_cp_mgr->consumer_list()) {
-            if (consumer) {
-                const auto pct = consumer->cp_progress_percent();
-                if (pct != 100) {
-                    consumer->repair_slow_cp();
-                    ++repair_attempted;
-                }
+            if (consumer && (consumer->cp_progress_percent() != 100)) {
+                consumer->repair_slow_cp();
+                ++repair_attempted;
             }
-            if (repair_attempted) { return; }
         }
-
-        HS_REL_ASSERT(0, "cp seems to be stuck. CP State={} total time elapsed {}", m_cp->to_string(),
-                      get_elapsed_time_ms(m_last_state_ch_time));
+        if (repair_attempted) {
+            ++m_repair_attempts;
+            LOGWARN("cp seems to be stuck, no progress for {} ms, repair attempt {}/{} on {} consumers. CP State={}",
+                    elapsed_ms, m_repair_attempts, max_repair_attempts, repair_attempted, m_cp->to_string());
+            // give the repairs time to take effect
+            m_last_repair_time = Clock::now();
+            return;
+        }
     }
+
+    HS_REL_ASSERT(0, "cp seems to be stuck, no progress for {} ms after {} repair attempts. CP State={}", elapsed_ms,
+                  m_repair_attempts, m_cp->to_string());
 }
 
 cp_id_t CPContext::id() const { return m_cp->id(); }

src/lib/device/journal_vdev.cpp

@@ -99,7 +99,10 @@ void JournalVirtualDev::init() {
     for (auto& [logdev_id, head] : logdev_head_map) {
         // Create descriptor for each logdev_id
         auto journal_desc = std::make_shared< JournalVirtualDev::Descriptor >(*this, logdev_id);
-        m_journal_descriptors.emplace(logdev_id, journal_desc);
+        {
+            std::lock_guard lock{m_mutex};
+            m_journal_descriptors.emplace(logdev_id, journal_desc);
+        }
         LOGINFOMOD(journalvdev, "Loading descriptor log_dev={}", logdev_id);
         // Traverse the list starting from the head and add those chunks
         // in order to the journal descriptor. next_chunk is stored in private_data.
@@ -178,6 +181,8 @@ uint64_t JournalVirtualDev::get_end_of_chunk(shared< Chunk >& chunk) const {
 }
 
 shared< JournalVirtualDev::Descriptor > JournalVirtualDev::open(logdev_id_t logdev_id) {
+    // Descriptors can be opened from several threads at once (VDevJournalIOTest.LongRunning opens one per thread)
+    std::lock_guard lock{m_mutex};
     auto it = m_journal_descriptors.find(logdev_id);
     if (it == m_journal_descriptors.end()) {
         auto journal_desc = std::make_shared< JournalVirtualDev::Descriptor >(*this, logdev_id);
@@ -194,6 +199,7 @@ shared< JournalVirtualDev::Descriptor > JournalVirtualDev::open(logdev_id_t logd
 }
 
 void JournalVirtualDev::destroy(logdev_id_t logdev_id) {
+    std::lock_guard lock{m_mutex};
     auto it = m_journal_descriptors.find(logdev_id);
     if (it == m_journal_descriptors.end()) {
         LOGERROR("logdev not found log_dev={}", logdev_id);

src/lib/logstore/log_store.cpp

@@ -214,6 +214,16 @@ bool home_log_store::truncate(logstore_seq_num_t upto_lsn, bool in_memory_trunca
                           m_start_lsn.load(std::memory_order_relaxed));
         return false;
     }
+    // Truncates of this store must not interleave: another one can move m_start_lsn and m_records past upto_lsn
+    // between the range check and the record lookup below. Lock order is m_truncate_mtx, then the logdev's flush guard
+    // and store map lock (taken by flush() and m_logdev->truncate()). LogDev::start() calls truncate() while holding
+    // the store map lock, but only up to the already truncated lsn, which returns above without taking m_truncate_mtx.
+    std::lock_guard truncate_lg{m_truncate_mtx};
+    if (upto_lsn < m_start_lsn) {
+        // a concurrent truncate got here first
+        decr_pending_request_num();
+        return false;
+    }
     flush();
 #ifndef NDEBUG
     auto cs = get_contiguous_completed_seq_num(0);

src/lib/replication/repl_dev/raft_repl_dev.cpp

@@ -1383,6 +1383,17 @@ bool RaftReplDev::wait_for_data_receive(std::vector< repl_req_ptr_t > const& rre
 }
 
 void RaftReplDev::check_and_fetch_remote_data(std::vector< repl_req_ptr_t > rreqs) {
+    // Same gate as on_fetch_data_received, on the sending side: once this dev is no longer ACTIVE, the reaper calls
+    // leave_group(), which frees the raft server that group_msg_service() resolves the originator through. Register
+    // the request before checking, so the reaper's stop() either waits for it or this check sees is_stopping().
+    incr_pending_request_num();
+    if (is_stopping() || (get_stage() != repl_dev_stage_t::ACTIVE)) {
+        RD_LOGD(NO_TRACE_ID, "Data Channel: not fetching data while not active (stopping={} stage={})", is_stopping(),
+                static_cast< int >(get_stage()));
+        decr_pending_request_num();
+        return;
+    }
+
     auto total_size_to_fetch = 0ul;
     std::vector< repl_req_ptr_t > next_batch_rreqs;
     auto const max_batch_size = HS_DYNAMIC_CONFIG(consensus.data_fetch_max_size_kb) * 1024ull;
@@ -1415,6 +1426,7 @@ void RaftReplDev::check_and_fetch_remote_data(std::vector< repl_req_ptr_t > rreq
         next_batch_rreqs.emplace_back(rreq);
     }
     fetch_data_from_remote(std::move(next_batch_rreqs));
+    decr_pending_request_num();
 }
 
 void RaftReplDev::fetch_data_from_remote(std::vector< repl_req_ptr_t > rreqs) {
@@ -1458,6 +1470,7 @@ void RaftReplDev::fetch_data_from_remote(std::vector< repl_req_ptr_t > rreqs) {
     // Fetch is fire-and-forget: detach() starts the coroutine and returns. Copy originator out of
     // rreqs.front() before rreqs is moved into the coroutine frame -- the reference would otherwise dangle.
     nuraft_mesg::svr_id_t const originator_id = originator;
+    incr_pending_request_num(); // released by fetch_data_coro once the fetch is fully handled
     sisl::async::detach(fetch_data_coro(std::move(builder), originator_id, std::move(rreqs)));
 }
 
@@ -1498,6 +1511,7 @@ sisl::async::task< void > RaftReplDev::fetch_data_coro(shared< flatbuffers::Flat
             }
         }
         COUNTER_INCREMENT(m_metrics, fetch_err_cnt, 1);
+        decr_pending_request_num();
         co_return;
     }
 
@@ -1506,6 +1520,7 @@ sisl::async::task< void > RaftReplDev::fetch_data_coro(shared< flatbuffers::Flat
     iomanager.run_on_forget(iomgr::reactor_regex::random_worker,
                             [this, r = std::move(response.value()), rreqs = std::move(rreqs)]() {
                                 handle_fetch_data_response(std::move(r), std::move(rreqs));
+                                decr_pending_request_num();
                             });
     co_return;
 }

src/lib/replication/service/raft_repl_service.cpp

@@ -739,7 +739,7 @@ void RaftReplService::gc_repl_devs() {
         return;
     }
 
-    std::vector< group_id_t > groups_to_leave;
+    std::vector< shared< RaftReplDev > > rdevs_to_leave;
     {
         std::shared_lock lg(m_rd_map_mtx);
         for (auto it = m_rd_map.begin(); it != m_rd_map.end(); ++it) {
@@ -750,15 +750,19 @@ void RaftReplService::gc_repl_devs() {
                 LOGINFOMOD(replication,
                            "repl_dev group_id={} was destroyed, shutting down the raft group in delayed fashion now",
                            rdev->group_id());
-                groups_to_leave.push_back(rdev->group_id());
+                rdevs_to_leave.push_back(rdev);
             }
         }
     }
 
     // Call leave_group to shut down the raft server and destroy all resources on the repl dev.
     // This operation may require acquiring the m_rd_map_mtx lock for some steps (e.g., trigger cp flush).
     // Therefore, we perform it outside the lock scope and then remove group from m_rd_map.
-    for (const auto& group_id : groups_to_leave) {
+    for (auto const& rdev : rdevs_to_leave) {
+        auto const group_id = rdev->group_id();
+        // Wait out in-flight requests on this dev (e.g. data fetches, which resolve peers through the raft server)
+        // before leave_group() frees the raft server.
+        rdev->stop();
         m_msg_mgr->leave_group(group_id);
         // notify consumer to cleanup any resources associated with the listener itself;
         m_repl_app->destroy_repl_dev_listener(group_id);