test: green the sanitizer lane — scale TIMEOUTs under sanitizers; feed rtl_capture_worker_test in lockstep. Principle VIII.

BUILD-OPTIONS.md

@@ -159,6 +159,7 @@ Additional cache values accept a value rather than ON/OFF:
 | `DEEPFIST_MODEL_BASE_URL` | Empty | Published, versioned HTTPS directory for the exact DeepFist assets. See the distribution prerequisite in [the DeepFist guide](docs/deepfist-cw-backend.md). |
 | `RADE_TAP_DIR` | `<build-directory>/rade_taps` | Directory for RADE WAV diagnostics; available when RADE and its taps are enabled. |
 | `AETHER_TEST_FFTW_TIMELIMIT` | `0.001` | Seconds FFTW may spend measuring each plan under test; an empty value allows unbounded measurement. |
+| `AETHER_SANITIZER_TIMEOUT_SCALE` | `4` | Positive integer. Multiplier applied to every test `TIMEOUT` when the build is sanitizer-instrumented (`AETHERSDR_SANITIZER` set, or `-fsanitize=` in the global C/C++ flags or in any configuration's `CMAKE_<LANG>_FLAGS_<CONFIG>`). An uninstrumented build keeps every limit exactly as written. |
 
 ## Keeping this reference current
 

tests/RtlInjectedDevice.h

@@ -1,10 +1,14 @@
 #pragma once
 #include "core/backends/rtl/RtlSdrWorker.h"
+#include <algorithm>
 #include <array>
 #include <atomic>
+#include <chrono>
 #include <condition_variable>
+#include <cstdlib>
 #include <memory>
 #include <mutex>
+#include <thread>
 
 namespace AetherSDR::test {
 using T = rtl::RtlCaptureTransaction;
@@ -82,6 +86,7 @@ class InjectedDevice final : public rtl::RtlSdrWorker::Device {
             --m_state->blocks;
             lock.unlock();
             iq.fill(m_state->iqLevel.load());
+            if (m_callbackDelay.count() > 0) { std::this_thread::sleep_for(m_callbackDelay); }
             callback(iq.data(), m_state->callbackBytes.load(), context);
             ++m_state->callbacks;
             lock.lock();
@@ -97,5 +102,13 @@ class InjectedDevice final : public rtl::RtlSdrWorker::Device {
     }
 private:
     std::shared_ptr<DeviceState> m_state;
+    // AETHER_TEST_RTL_CALLBACK_DELAY_MS makes every callback slow, the way a
+    // sanitizer build's DSP is, so a test that silently assumes a fast device
+    // thread fails deterministically instead of only under TSan.
+    // rtl_capture_worker_slow_device_test runs with it set.
+    std::chrono::milliseconds m_callbackDelay{[] {
+        const char* value = std::getenv("AETHER_TEST_RTL_CALLBACK_DELAY_MS");
+        return value ? std::max(0, std::atoi(value)) : 0;
+    }()};
 };
 } // namespace AetherSDR::test

tests/rtl_capture_worker_test.cpp

@@ -50,10 +50,10 @@ static void check(bool value, const char* message)
 }
 using DeviceState = AetherSDR::test::DeviceState;
 using InjectedDevice = AetherSDR::test::InjectedDevice;
-template<class Predicate> bool waitFor(Predicate predicate)
+template<class Predicate> bool waitFor(Predicate predicate, qint64 boundMs = 3000)
 {
     QElapsedTimer timer; timer.start();
-    while (!predicate() && timer.elapsed() < 3000) {
+    while (!predicate() && timer.elapsed() < boundMs) {
         QCoreApplication::processEvents(); QThread::msleep(1);
     }
     return predicate();
@@ -179,49 +179,60 @@ static void receiverTuneKeepsDisplayAverage()
             ++frames;
             std::memcpy(&lastDc, frame.constData() + frame.size() / 2, sizeof(lastDc));
         });
-    for (int i = 0; i < 8; ++i) { device->block(); }
-    check(waitFor([&] { return frames > 0; }), "averaging worker seeds a complete observation");
+    // Every step below asserts on the FIRST window after an adoption, so no
+    // block may still be queued when a change adopts: a queued block would
+    // join the post-adoption window and shift which window the next frame
+    // is. block() only queues; the device thread consumes on its own time,
+    // which a sanitizer build slows enough to leave a backlog. So feed in
+    // lockstep, one consumed block at a time.
+    const auto feedOne = [&] {
+        const int before = device->callbacks;
+        device->block();
+        return waitFor([&] { return device->callbacks > before; });
+    };
+    // Adopt a change at a real callback boundary: one consumed block per pass
+    // until the backend has published the new revision.
+    const auto adopt = [&] {
+        return waitFor([&] { feedOne(); return !rtl::RtlCaptureBackendTestAccess::busy(receiver); });
+    };
+    // The first accepted frame after the latest adoption, one block at a time.
+    const auto nextFrame = [&] {
+        const int before = frames;
+        return waitFor([&] {
+            if (frames > before) { return true; }
+            feedOne(); QCoreApplication::processEvents();
+            return frames > before;
+        });
+    };
+    check(nextFrame(), "averaging worker seeds a complete observation");
     const int writes = device->writes;
     device->iqLevel = 190;
     receiver.setSliceFrequency(0, 100'100'000);
-    check(waitFor([&] { device->block(); return !rtl::RtlCaptureBackendTestAccess::busy(receiver); }),
-          "receiver-only tune adopts through the actual USB callback");
-    const int beforeFrames = frames;
-    for (int i = 0; i < 8; ++i) { device->block(); }
-    check(waitFor([&] { return frames > beforeFrames; }), "tuned receiver emits new accepted FFT");
+    check(adopt(), "receiver-only tune adopts through the actual USB callback");
+    check(nextFrame(), "tuned receiver emits new accepted FFT");
     const double raw = 20 * std::log10(std::sqrt(2.) * (190 - 127.5) / 127.5
                                       * .35875 * (65535. / 65536));
     check(device->writes == writes && lastDc < raw - 8,
           "receiver-only adoption preserves display smoothing with unchanged hardware");
     receiver.setPanRfGain({}, 25);
-    check(waitFor([&] { device->block(); return !rtl::RtlCaptureBackendTestAccess::busy(receiver); }),
-          "gain change adopts through verified hardware path");
-    const int beforeGainFrames = frames;
-    for (int i = 0; i < 8; ++i) { device->block(); }
-    check(waitFor([&] { return frames > beforeGainFrames; }), "new gain emits complete frame");
+    check(adopt(), "gain change adopts through verified hardware path");
+    check(nextFrame(), "new gain emits complete frame");
     check(std::abs(lastDc - raw) < .02,
           "gain transition discards incompatible amplitude history");
     device->iqLevel = 130;
     receiver.invokeExtension("rtl", "ppm.set", 1, 1);
     check(waitFor([&] { return !rtl::RtlCaptureBackendTestAccess::busy(receiver); }),
           "PPM change completes its hardware readback");
-    const int beforePpmFrames = frames;
-    for (int i = 0; i < 8; ++i) { device->block(); }
-    check(waitFor([&] { return frames > beforePpmFrames; }), "PPM change emits complete observation");
+    check(nextFrame(), "PPM change emits complete observation");
     const double weak = 20 * std::log10(std::sqrt(2.) * (130 - 127.5) / 127.5
                                        * .35875 * (65535. / 65536));
     check(rtl::RtlCaptureBackendTestAccess::state(receiver).hardware.ppm == 1
         && std::abs(lastDc - weak) < .02,
           "PPM change cannot reinterpret the previous RF history");
     device->iqLevel = 190;
     receiver.invokeExtension("rtl", "dc_suppression.set", 2, true);
-    check(waitFor([&] {
-        device->block(); QThread::msleep(20);
-        return !rtl::RtlCaptureBackendTestAccess::busy(receiver);
-    }), "DC correction adopts on a real callback boundary");
-    const int beforeDcFrames = frames;
-    for (int i = 0; i < 8; ++i) { device->block(); }
-    check(waitFor([&] { return frames > beforeDcFrames; }), "DC correction emits complete observation");
+    check(adopt(), "DC correction adopts on a real callback boundary");
+    check(nextFrame(), "DC correction emits complete observation");
     const double pole = std::exp(-2 * std::numbers::pi * rtl::RtlDcBlocker::kCornerHz / 2'400'000);
     double correctedWindowGain = 0;
     for (int i = 0; i < 65536; ++i) {
@@ -527,7 +538,12 @@ int main(int argc, char** argv)
         check(!activeFrame.current(), "park revokes the previously published native PCM stream");
         const int audioAtPark = sliceAudioPackets;
         const int iqSpectraAtPark = spectra;
+        const int parkedCallbacks = device->callbacks + 64;
         for (int i = 0; i < 64; ++i) { pump(); QCoreApplication::processEvents(); }
+        // pump() only queues; both checks below are about blocks the worker
+        // actually processed, so wait for the queue to drain first.
+        check(waitFor([&] { return device->callbacks >= parkedCallbacks; }, 30000),
+              "parked IQ callbacks are all processed");
         QThread::msleep(20); QCoreApplication::processEvents();
         check(sliceAudioPackets == audioAtPark,
             "IQ callbacks with every slice parked publish no slice PCM");
@@ -559,8 +575,11 @@ int main(int argc, char** argv)
         // produces real FM packets. This is queue saturation, not a fake counter.
         const int targetCallbacks = device->callbacks + 200;
         { std::lock_guard lock(device->mutex); device->blocks += 200; device->changed.notify_all(); }
+        // Bounded only so a broken worker fails rather than hangs: 200 callbacks
+        // take well under 3 s normally, but a sanitizer build's DSP is several
+        // times slower and must not read as a missing saturation.
         QElapsedTimer stalledOwner; stalledOwner.start();
-        while (device->callbacks < targetCallbacks && stalledOwner.elapsed() < 3000) { QThread::msleep(1); }
+        while (device->callbacks < targetCallbacks && stalledOwner.elapsed() < 30000) { QThread::msleep(1); }
         check(device->callbacks >= targetCallbacks, "injected acquisition reaches bounded queue saturation");
         check(waitFor([&] {
             return multiple.healthSnapshot().values.value("rtlQueueDrops").toULongLong() > 0;

tests/tests.cmake

@@ -1638,6 +1638,13 @@ if(AETHER_BACKEND_RTL)
     target_link_libraries(rtl_capture_worker_test PRIVATE aethercore Qt6::Core Qt6::Test)
     add_test(NAME rtl_capture_worker_test COMMAND rtl_capture_worker_test)
     set_tests_properties(rtl_capture_worker_test PROPERTIES TIMEOUT 20)
+    # The same binary with every injected USB callback 30 ms slow: a fast
+    # device thread is no longer something a check can assume by accident.
+    # 30 ms is where the old fixed-count block feeding first broke the
+    # DC-correction check, which is how it failed under TSan (#6202).
+    add_test(NAME rtl_capture_worker_slow_device_test COMMAND rtl_capture_worker_test)
+    set_tests_properties(rtl_capture_worker_slow_device_test PROPERTIES
+        TIMEOUT 60 ENVIRONMENT "AETHER_TEST_RTL_CALLBACK_DELAY_MS=30")
 
     # Generated IQ only: genuine resolution, continuity, gain and detector independence.
     add_executable(rtl_spectrum_resolution_test tests/rtl_spectrum_resolution_test.cpp)
@@ -7963,6 +7970,46 @@ target_compile_definitions(aether_test_wisdom_isolation PRIVATE
 # property always beats a `ctest --timeout` flag, so this is authoritative
 # in every lane — gate steps, sanitizers, and local dev alike. Applied by
 # aether_retrofit_tests() below.
+#
+# ── Sanitizer builds scale every TIMEOUT ────────────────────────────────────
+#
+# The ceilings above, and each test's own TIMEOUT, are sized for an
+# uninstrumented build; sanitizer instrumentation makes the same test several
+# times slower in wall-clock time. So in an instrumented build,
+# aether_retrofit_tests() multiplies every TIMEOUT (the test's own, or the
+# default above) by AETHER_SANITIZER_TIMEOUT_SCALE. An ordinary build keeps
+# every limit exactly as written, so a slow test still fails fast there; if one
+# outgrows its limit in an ordinary build, raise ITS TIMEOUT, not the scale.
+#
+# "Instrumented": AETHERSDR_SANITIZER is set, or -fsanitize= appears in the
+# global C/C++ flags (how sanitizers.yml builds) or in a configuration's
+# CMAKE_<LANG>_FLAGS_<CONFIG>. A test TIMEOUT is not per-configuration, so a
+# multi-config generator scales when ANY of its configurations is instrumented.
+set(AETHER_SANITIZER_TIMEOUT_SCALE 4 CACHE STRING
+    "Multiplier applied to every test TIMEOUT in a sanitizer-instrumented build")
+if(NOT AETHER_SANITIZER_TIMEOUT_SCALE MATCHES "^[1-9][0-9]*$")
+    message(FATAL_ERROR
+        "AETHER_SANITIZER_TIMEOUT_SCALE must be a positive integer, got "
+        "'${AETHER_SANITIZER_TIMEOUT_SCALE}'.")
+endif()
+set(_aether_compile_flags "${CMAKE_C_FLAGS} ${CMAKE_CXX_FLAGS}")
+if(CMAKE_CONFIGURATION_TYPES)
+    set(_aether_flag_configs ${CMAKE_CONFIGURATION_TYPES})
+else()
+    set(_aether_flag_configs ${CMAKE_BUILD_TYPE})
+endif()
+foreach(_aether_flag_config IN LISTS _aether_flag_configs)
+    string(TOUPPER "${_aether_flag_config}" _aether_flag_config)
+    string(APPEND _aether_compile_flags
+        " ${CMAKE_C_FLAGS_${_aether_flag_config}} ${CMAKE_CXX_FLAGS_${_aether_flag_config}}")
+endforeach()
+set(_aether_tests_instrumented OFF)
+if(NOT AETHERSDR_SANITIZER STREQUAL "none"
+        OR _aether_compile_flags MATCHES "-fsanitize=")
+    set(_aether_tests_instrumented ON)
+    message(STATUS "Sanitizer-instrumented build: test TIMEOUTs scaled x"
+                   "${AETHER_SANITIZER_TIMEOUT_SCALE}")
+endif()
 
 # Every executable declared in this file: the directory's targets now, minus the
 # baseline from the top of the file, each one checked against the names this
@@ -8053,8 +8100,21 @@ function(aether_retrofit_tests)
         # replaced too: the rule is a ceiling on every test.
         get_test_property(${_test} TIMEOUT _existing_timeout)
         if(NOT _existing_timeout)
+            set(_existing_timeout 300)
             set_tests_properties(${_test} PROPERTIES TIMEOUT 300)
         endif()
+
+        # The sanitizer allowance (see "Sanitizer builds scale every TIMEOUT").
+        if(_aether_tests_instrumented)
+            if(NOT _existing_timeout MATCHES "^[0-9]+$")
+                message(FATAL_ERROR
+                    "test ${_test}: TIMEOUT '${_existing_timeout}' is not a whole "
+                    "number of seconds, so it cannot be scaled for a sanitizer build.")
+            endif()
+            math(EXPR _scaled_timeout
+                "${_existing_timeout} * ${AETHER_SANITIZER_TIMEOUT_SCALE}")
+            set_tests_properties(${_test} PROPERTIES TIMEOUT ${_scaled_timeout})
+        endif()
     endforeach()
 
     cmake_language(DEFER CALL aether_assert_tests_retrofitted)