Stabilize timing-sensitive tests

lib/offline/KillSwitchManager.hpp

@@ -7,11 +7,14 @@
 
 #include "pal/PAL.hpp"
 
+#include <functional>
+#include <limits>
 #include <list>
 #include <map>
 #include <mutex>
 #include <stdexcept>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include <atomic>
@@ -21,13 +24,24 @@ namespace MAT_NS_BEGIN {
     class KillSwitchManager
     {
     public:
+        using Clock = std::function<int64_t()>;
 
         bool isActive()
         {
             return !m_tokenTime.empty();
         }
 
-        KillSwitchManager() : m_isRetryAfterActive(false), m_retryAfterExpiryTime(0)
+        KillSwitchManager()
+            : KillSwitchManager([]() { return static_cast<int64_t>(PAL::getMonotonicTimeMs()); })
+        {
+        }
+
+        explicit KillSwitchManager(Clock clock)
+            : m_clock(clock
+                ? std::move(clock)
+                : Clock([]() { return static_cast<int64_t>(PAL::getMonotonicTimeMs()); })),
+              m_isRetryAfterActive(false),
+              m_retryAfterExpiryTime(0)
         {
         }
 
@@ -45,8 +59,9 @@ namespace MAT_NS_BEGIN {
                 int64_t timeinSecs = 0;
                 if (tryParseSeconds(timeStr, timeinSecs) && timeinSecs > 0)
                 {
+                    const int64_t expiryTime = expiryFromNow(timeinSecs);
                     std::lock_guard<std::mutex> guard(m_lock);
-                    m_retryAfterExpiryTime = PAL::getUtcSystemTime() + timeinSecs;
+                    m_retryAfterExpiryTime = expiryTime;
                     m_isRetryAfterActive = true;
                 }
             }
@@ -101,20 +116,22 @@ namespace MAT_NS_BEGIN {
 
         void addToken(const std::string& tokenId, int64_t timeInSeconds)
         {
-            std::lock_guard<std::mutex> guard(m_lock);
             if (timeInSeconds > 0)
             {
-                m_tokenTime[tokenId] = PAL::getUtcSystemTime() + timeInSeconds; //convert milisec to sec
+                const int64_t expiryTime = expiryFromNow(timeInSeconds);
+                std::lock_guard<std::mutex> guard(m_lock);
+                m_tokenTime[tokenId] = expiryTime;
             }
         }
 
         bool isTokenBlocked(const std::string& tokenId)
         {
+            const int64_t now = m_clock();
             std::lock_guard<std::mutex> guard(m_lock);
 
             if (m_isRetryAfterActive)
             {
-                if (m_retryAfterExpiryTime > PAL::getUtcSystemTime())
+                if (m_retryAfterExpiryTime > now)
                 {
                     return true;//always return true for all tokens
                 }
@@ -129,7 +146,7 @@ namespace MAT_NS_BEGIN {
             {//found, check the time stamp
                 int64_t timeStamp = m_tokenTime[tokenId];
 
-                if (timeStamp > PAL::getUtcSystemTime())  //convert milisec to sec
+                if (timeStamp > now)
                 {
                     return true;
                 }
@@ -169,6 +186,24 @@ namespace MAT_NS_BEGIN {
         }
 
     private:
+        // Precondition: seconds > 0. All call sites enforce this (handleResponse
+        // and addToken both guard with `timeinSecs > 0` / `timeInSeconds > 0`).
+        // Passing a non-positive value is UB: a negative durationMs makes the
+        // overflow check `now > maxTime - durationMs` wrap (signed overflow), so
+        // the result is unpredictable — do not relax the call-site guards.
+        int64_t expiryFromNow(int64_t seconds) const
+        {
+            constexpr int64_t millisecondsPerSecond = 1000;
+            constexpr int64_t maxTime = std::numeric_limits<int64_t>::max();
+            const int64_t now = m_clock();
+            if (seconds > maxTime / millisecondsPerSecond)
+            {
+                return maxTime;
+            }
+            const int64_t durationMs = seconds * millisecondsPerSecond;
+            return now > maxTime - durationMs ? maxTime : now + durationMs;
+        }
+
         // Parse a count of seconds from a response-header value (Retry-After /
         // kill-duration). Returns false when the value is malformed or out of
         // range instead of letting std::stoll throw: the worker thread that drives
@@ -225,8 +260,8 @@ namespace MAT_NS_BEGIN {
                 // Either way the std::exception catch below ignores the value rather
                 // than crashing.
                 const long long parsed = std::stoll(value.substr(begin, end - begin));
-                // Clamp to a value that cannot overflow when later added to a current
-                // UTC timestamp (seconds) to compute an expiry time. No legitimate
+                // Clamp to a value that cannot overflow when later converted to
+                // milliseconds to compute an expiry time. No legitimate
                 // Retry-After / kill-duration approaches this; an absurd value is
                 // capped instead of wrapping the expiry into the past.
                 const int64_t kMaxSeconds = 100LL * 365 * 24 * 60 * 60; // ~100 years
@@ -272,6 +307,7 @@ namespace MAT_NS_BEGIN {
             return true;
         }
 
+        Clock           m_clock;
         std::map<std::string, int64_t> m_tokenTime;
         std::mutex      m_lock;
         bool            m_isRetryAfterActive;
@@ -280,4 +316,3 @@ namespace MAT_NS_BEGIN {
 
 } MAT_NS_END
 #endif
-

tests/functests/APITest.cpp

@@ -210,6 +210,98 @@ class TestDebugEventListener : public DebugEventListener {
     }
 };
 
+// Keep requests in flight until teardown cancels them, then simulate a connection
+// reset while honoring IHttpClient's exactly-once callback contract.
+class NetworkFailureHttpClient final : public IHttpClient
+{
+public:
+    IHttpRequest* CreateRequest() override
+    {
+        return new SimpleHttpRequest("bad-network-" + std::to_string(m_nextRequestId.fetch_add(1)));
+    }
+
+    void SendRequestAsync(IHttpRequest* request, IHttpResponseCallback* callback) override
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        m_pending[request->GetId()] = callback;
+        m_sent.fetch_add(1);
+    }
+
+    void CancelRequestAsync(const std::string& id) override
+    {
+        IHttpResponseCallback* callback = nullptr;
+        {
+            std::lock_guard<std::mutex> lock(m_mutex);
+            auto it = m_pending.find(id);
+            if (it != m_pending.end())
+            {
+                callback = it->second;
+                m_pending.erase(it);
+            }
+        }
+        if (callback != nullptr)
+        {
+            m_cancelled.fetch_add(1);
+            CompleteWithNetworkFailure(id, callback);
+        }
+    }
+
+    void CancelAllRequests() override
+    {
+        std::map<std::string, IHttpResponseCallback*> pending;
+        {
+            std::lock_guard<std::mutex> lock(m_mutex);
+            pending.swap(m_pending);
+        }
+        m_cancelled.fetch_add(static_cast<unsigned>(pending.size()));
+        for (const auto& request : pending)
+        {
+            CompleteWithNetworkFailure(request.first, request.second);
+        }
+    }
+
+    bool WaitForRequest(unsigned timeoutMs) const
+    {
+        const auto deadline = PAL::getMonotonicTimeMs() + timeoutMs;
+        while (SentCount() == 0 && PAL::getMonotonicTimeMs() < deadline)
+        {
+            PAL::sleep(10);
+        }
+        return SentCount() > 0;
+    }
+
+    unsigned SentCount() const
+    {
+        return m_sent.load();
+    }
+
+    unsigned CancelledCount() const
+    {
+        return m_cancelled.load();
+    }
+
+    unsigned CompletedCount() const
+    {
+        return m_completed.load();
+    }
+
+private:
+    void CompleteWithNetworkFailure(const std::string& id, IHttpResponseCallback* callback)
+    {
+        auto response = new SimpleHttpResponse("failure-" + id);
+        response->m_result = HttpResult_NetworkFailure;
+        callback->OnHttpResponse(response);
+        m_completed.fetch_add(1);
+    }
+
+    mutable std::mutex m_mutex;
+    std::map<std::string, IHttpResponseCallback*> m_pending;
+    std::atomic<unsigned> m_nextRequestId{0};
+    std::atomic<unsigned> m_sent{0};
+    std::atomic<unsigned> m_cancelled{0};
+    std::atomic<unsigned> m_completed{0};
+};
+
 /// <summary>
 /// Add all event listeners
 /// </summary>
@@ -1204,41 +1296,43 @@ TEST(APITest, LogConfiguration_MsRoot_Check)
 TEST(APITest, LogManager_BadNetwork_Test)
 {
     auto& config = LogManager::GetLogConfiguration();
-
-    // Clean temp file first
     const char *cacheFilePath = "bad-network.db";
     std::string fileName = MAT::GetTempDirectory();
     fileName += cacheFilePath;
-    printf("remove %s\n", fileName.c_str());
     std::remove(fileName.c_str());
     std::remove((fileName + "-wal").c_str());
     std::remove((fileName + "-shm").c_str());
     std::remove((fileName + "-journal").c_str());
 
-    for (auto url : {
-#if 0 /* [MG}: Temporary change to avoid GitHub Actions crash #92 */
-        "https://0.0.0.0/",
-        "https://127.0.0.1/",
-#endif
-        "https://mobile.events-sandbox.data.microsoft.com/OneCollector/1.0/",
-        "https://invalid.host.name.microsoft.com/"
-        })
-    {
-        printf("--- trying %s", url);
-        config[CFG_STR_CACHE_FILE_PATH] = cacheFilePath;
-        config[CFG_INT_TRACE_LEVEL_MASK] = 0;
-        config[CFG_INT_TRACE_LEVEL_MIN] = ACTTraceLevel_Warn;
-        config[CFG_INT_SDK_MODE] = SdkModeTypes::SdkModeTypes_CS;
-        config[CFG_INT_MAX_TEARDOWN_TIME] = 0;
-        config[CFG_STR_COLLECTOR_URL] = url;
-        size_t numIterations = 5;
-        while (numIterations--)
-        {
-            printf(".");
-            EXPECT_GE(StressSingleThreaded(config), MAX_ITERATIONS);
-        }
-        printf("\n");
-    }
+    auto httpClient = std::make_shared<NetworkFailureHttpClient>();
+    config.AddModule(CFG_MODULE_HTTP_CLIENT, httpClient);
+    config[CFG_STR_CACHE_FILE_PATH] = cacheFilePath;
+    config[CFG_INT_TRACE_LEVEL_MASK] = 0;
+    config[CFG_INT_TRACE_LEVEL_MIN] = ACTTraceLevel_Warn;
+    config[CFG_INT_SDK_MODE] = SdkModeTypes::SdkModeTypes_CS;
+    config[CFG_INT_MAX_TEARDOWN_TIME] = 0;
+    config[CFG_STR_COLLECTOR_URL] = "https://unused.invalid/";
+
+    TestDebugEventListener debugListener;
+    addAllListeners(debugListener);
+    LogManager::AddEventListener(DebugEventType::EVT_HTTP_FAILURE, debugListener);
+    auto logger = LogManager::Initialize(TEST_TOKEN, config);
+    LogManager::SetTransmitProfile(TransmitProfile_RealTime);
+    logger->LogEvent("badNetworkEvent");
+    LogManager::UploadNow();
+
+    const bool requestStarted = httpClient->WaitForRequest(10000);
+    LogManager::FlushAndTeardown();
+    LogManager::RemoveEventListener(DebugEventType::EVT_HTTP_FAILURE, debugListener);
+    removeAllListeners(debugListener);
+    config.AddModule(CFG_MODULE_HTTP_CLIENT, nullptr);
+
+    EXPECT_TRUE(requestStarted);
+    EXPECT_GE(debugListener.numLogged.load(), 1u);
+    EXPECT_GE(debugListener.numHttpError.load(), 1u);
+    EXPECT_GE(httpClient->SentCount(), 1u);
+    EXPECT_EQ(httpClient->SentCount(), httpClient->CancelledCount());
+    EXPECT_EQ(httpClient->CancelledCount(), httpClient->CompletedCount());
 }
 
 TEST(APITest, LogManager_GetLoggerSameLoggerMultithreaded)
@@ -1485,4 +1579,3 @@ TEST(APITest, Custom_Decorator)
 #endif // HAVE_MAT_DEFAULT_HTTP_CLIENT
 
 // TEST_PULL_ME_IN(APITest)
-