diff --git a/lib/offline/KillSwitchManager.hpp b/lib/offline/KillSwitchManager.hpp index a70569877..d5f5a1211 100644 --- a/lib/offline/KillSwitchManager.hpp +++ b/lib/offline/KillSwitchManager.hpp @@ -7,11 +7,14 @@ #include "pal/PAL.hpp" +#include +#include #include #include #include #include #include +#include #include #include @@ -21,13 +24,24 @@ namespace MAT_NS_BEGIN { class KillSwitchManager { public: + using Clock = std::function; bool isActive() { return !m_tokenTime.empty(); } - KillSwitchManager() : m_isRetryAfterActive(false), m_retryAfterExpiryTime(0) + KillSwitchManager() + : KillSwitchManager([]() { return static_cast(PAL::getMonotonicTimeMs()); }) + { + } + + explicit KillSwitchManager(Clock clock) + : m_clock(clock + ? std::move(clock) + : Clock([]() { return static_cast(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 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 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 guard(m_lock); + m_tokenTime[tokenId] = expiryTime; } } bool isTokenBlocked(const std::string& tokenId) { + const int64_t now = m_clock(); std::lock_guard 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::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 m_tokenTime; std::mutex m_lock; bool m_isRetryAfterActive; @@ -280,4 +316,3 @@ namespace MAT_NS_BEGIN { } MAT_NS_END #endif - diff --git a/tests/functests/APITest.cpp b/tests/functests/APITest.cpp index 0347807f6..baea0112e 100644 --- a/tests/functests/APITest.cpp +++ b/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 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 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 pending; + { + std::lock_guard lock(m_mutex); + pending.swap(m_pending); + } + m_cancelled.fetch_add(static_cast(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 m_pending; + std::atomic m_nextRequestId{0}; + std::atomic m_sent{0}; + std::atomic m_cancelled{0}; + std::atomic m_completed{0}; +}; + /// /// Add all event listeners /// @@ -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(); + 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) - diff --git a/tests/functests/BasicFuncTests.cpp b/tests/functests/BasicFuncTests.cpp index 438411425..bc879d3e6 100644 --- a/tests/functests/BasicFuncTests.cpp +++ b/tests/functests/BasicFuncTests.cpp @@ -541,6 +541,35 @@ class BasicFuncTests : public ::testing::Test, } return result; } + + bool waitForEvent(const std::string& name, unsigned timeoutMs, size_t& nextRequestIndex) + { + const auto deadline = PAL::getMonotonicTimeMs() + timeoutMs; + while (PAL::getMonotonicTimeMs() < deadline) + { + std::vector newRequests; + { + LOCKGUARD(mtx_requests); + while (nextRequestIndex < receivedRequests.size()) + { + newRequests.push_back(receivedRequests[nextRequestIndex]); + ++nextRequestIndex; + } + } + for (const auto& request : newRequests) + { + for (const auto& record : decodeRequest(request, false)) + { + if (record.name == name) + { + return true; + } + } + } + PAL::sleep(10); + } + return false; + } }; @@ -1110,6 +1139,17 @@ public : break; }; } + + bool waitForAtLeast(const std::atomic& counter, unsigned expected, unsigned timeoutMs) + { + const auto deadline = PAL::getMonotonicTimeMs() + timeoutMs; + while (counter.load() < expected && PAL::getMonotonicTimeMs() < deadline) + { + PAL::sleep(10); + } + return counter.load() >= expected; + } + void printStats(){ std::cerr << "[ ] numLogged = " << numLogged << std::endl; std::cerr << "[ ] numSent = " << numSent << std::endl; @@ -1231,84 +1271,71 @@ TEST_F(BasicFuncTests, killSwitchWorks) TEST_F(BasicFuncTests, killIsTemporary) { CleanStorage(); - // Create the configuration to send to fake server auto configuration = LogManager::GetLogConfiguration(); configuration[CFG_INT_TRACE_LEVEL_MASK] = 0xFFFFFFFF; configuration[CFG_INT_TRACE_LEVEL_MIN] = ACTTraceLevel_Warn; configuration[CFG_INT_SDK_MODE] = SdkModeTypes::SdkModeTypes_CS; - configuration[CFG_INT_RAM_QUEUE_SIZE] = 4096 * 20; configuration[CFG_STR_CACHE_FILE_PATH] = TEST_STORAGE_FILENAME; - configuration[CFG_INT_MAX_TEARDOWN_TIME] = 2; // 2 seconds wait on shutdown + configuration[CFG_INT_MAX_TEARDOWN_TIME] = 2; configuration[CFG_STR_COLLECTOR_URL] = serverAddress.c_str(); - configuration[CFG_MAP_HTTP][CFG_BOOL_HTTP_COMPRESSION] = false; // disable compression for now - configuration[CFG_MAP_METASTATS_CONFIG]["interval"] = 30 * 60; // 30 mins - configuration[CFG_MAP_METASTATS_CONFIG]["enabled"] = true; // opt in to stats (disabled by default since #1420) - + configuration[CFG_MAP_HTTP][CFG_BOOL_HTTP_COMPRESSION] = false; + configuration[CFG_MAP_METASTATS_CONFIG]["interval"] = 30 * 60; + configuration[CFG_MAP_METASTATS_CONFIG]["enabled"] = true; configuration["name"] = __FILE__; configuration["version"] = "1.0.0"; - configuration["config"] = { { "host", __FILE__ } }; // Host instance + configuration["config"] = { { "host", __FILE__ } }; - // set the killed token on the server - server.setKilledToken(KILLED_TOKEN, 10); + constexpr unsigned killDurationSec = 5; + server.setKilledToken(KILLED_TOKEN, killDurationSec); KillSwitchListener listener; addListeners(listener); - // Log 100 events from valid and invalid 4 times - int repetitions = 4; - for (int i = 0; i < repetitions; i++) { - // Initialize the logger for the valid token and log 100 events - LogManager::Initialize(TEST_TOKEN, configuration); - LogManager::ResumeTransmission(); - auto myLogger = LogManager::GetLogger(TEST_TOKEN, "killed"); - int numIterations = 100; - while (numIterations--) { - EventProperties event1("fooEvent"); - event1.SetProperty("property", "value"); - myLogger->LogEvent(event1); - } - // Initialize the logger for the killed token and log 100 events - LogManager::Initialize(KILLED_TOKEN, configuration); - LogManager::ResumeTransmission(); - myLogger = LogManager::GetLogger(KILLED_TOKEN, "killed"); - numIterations = 100; - while (numIterations--) { - EventProperties event2("failEvent"); - event2.SetProperty("property", "value"); - myLogger->LogEvent(event2); - } - } - // Try and wait to upload - LogManager::UploadNow(); - PAL::sleep(2000); - // Sleep for 11 seconds so the killed time has expired, clear the killed tokens on server - PAL::sleep(11000); - server.clearKilledTokens(); - // Log 100 events with valid logger - LogManager::Initialize(TEST_TOKEN, configuration); - LogManager::ResumeTransmission(); - auto myLogger = LogManager::GetLogger(TEST_TOKEN, "killed"); - int numIterations = 100; - while (numIterations--) { - EventProperties event1("fooEvent"); - event1.SetProperty("property", "value"); - myLogger->LogEvent(event1); - } LogManager::Initialize(KILLED_TOKEN, configuration); + LogManager::SetTransmitProfile(TransmitProfile_RealTime); LogManager::ResumeTransmission(); - myLogger = LogManager::GetLogger(KILLED_TOKEN, "killed"); - numIterations = 100; - while (numIterations--) { - EventProperties event2("failEvent"); - event2.SetProperty("property", "value"); - myLogger->LogEvent(event2); + + auto killedLogger = LogManager::GetLogger(KILLED_TOKEN, "killed"); + killedLogger->LogEvent("activateKillSwitch"); + LogManager::UploadNow(); + + const bool killSwitchActivated = listener.waitForAtLeast(listener.numHttpOK, 1, 10000); + if (!killSwitchActivated) + { + LogManager::FlushAndTeardown(); + removeListeners(listener); + server.clearKilledTokens(); } - // Expect to 0 events to be dropped - EXPECT_EQ(uint32_t { 0 }, listener.numDropped); - LogManager::FlushAndTeardown(); + ASSERT_TRUE(killSwitchActivated) << "Kill-switch response was not observed before timeout"; + server.clearKilledTokens(); - listener.printStats(); + const unsigned droppedBeforeKill = listener.numDropped.load(); + const auto activeDeadline = PAL::getMonotonicTimeMs() + 2000; + unsigned probe = 0; + while (listener.numDropped.load() == droppedBeforeKill + && PAL::getMonotonicTimeMs() < activeDeadline) + { + killedLogger->LogEvent("blockedWhileKillIsActive" + std::to_string(probe++)); + PAL::sleep(20); + } + EXPECT_GT(listener.numDropped.load(), droppedBeforeKill); + + // Poll until the kill-switch TTL expires and the SDK resumes sending. + // Budget: kill duration + 5 s headroom; the extra 100 ms absorbs any + // request that was dispatched just before the deadline fires. + const auto expiryDeadline = PAL::getMonotonicTimeMs() + (killDurationSec + 5) * 1000 + 100; + size_t nextRequestIndex = 0; + bool acceptedAfterKillExpires = false; + while (!acceptedAfterKillExpires && PAL::getMonotonicTimeMs() < expiryDeadline) + { + killedLogger->LogEvent("acceptedAfterKillExpires"); + LogManager::UploadNow(); + acceptedAfterKillExpires = waitForEvent("acceptedAfterKillExpires", 100, nextRequestIndex); + } + EXPECT_TRUE(acceptedAfterKillExpires); + + LogManager::FlushAndTeardown(); removeListeners(listener); server.clearKilledTokens(); } diff --git a/tests/unittests/KillSwitchManagerTests.cpp b/tests/unittests/KillSwitchManagerTests.cpp index 15aaaee18..ceec1f450 100644 --- a/tests/unittests/KillSwitchManagerTests.cpp +++ b/tests/unittests/KillSwitchManagerTests.cpp @@ -16,6 +16,34 @@ TEST(KillSwitchManagerTests, handleResponse_ValidRetryAfter_ActivatesRetryAfter) ASSERT_TRUE(manager.isRetryAfterActive()); } +TEST(KillSwitchManagerTests, constructor_EmptyClockUsesMonotonicClock) +{ + KillSwitchManager manager(KillSwitchManager::Clock{}); + HttpHeaders headers; + headers.add("Retry-After", "120"); + + ASSERT_NO_THROW(manager.handleResponse(headers)); + EXPECT_TRUE(manager.isTokenBlocked("any-token")); +} + +TEST(KillSwitchManagerTests, handleResponse_RetryAfterExpiresAtDeadline) +{ + int64_t nowMs = 1000; + KillSwitchManager manager([&nowMs]() { return nowMs; }); + HttpHeaders headers; + headers.add("Retry-After", "120"); + + manager.handleResponse(headers); + ASSERT_TRUE(manager.isTokenBlocked("any-token")); + + nowMs += 119999; + EXPECT_TRUE(manager.isTokenBlocked("any-token")); + + nowMs += 1; + EXPECT_FALSE(manager.isTokenBlocked("any-token")); + EXPECT_FALSE(manager.isRetryAfterActive()); +} + TEST(KillSwitchManagerTests, handleResponse_NonNumericRetryAfter_DoesNotThrowAndIsIgnored) { KillSwitchManager manager; @@ -120,6 +148,25 @@ TEST(KillSwitchManagerTests, handleResponse_ValidKillTokenAndDuration_BlocksToke ASSERT_TRUE(manager.isTokenBlocked("tenant-token-1")); } +TEST(KillSwitchManagerTests, handleResponse_KillDurationExpiresAtDeadline) +{ + int64_t nowMs = 1000; + KillSwitchManager manager([&nowMs]() { return nowMs; }); + HttpHeaders headers; + headers.add("kill-tokens", "tenant-token-1"); + headers.add("kill-duration", "10"); + + ASSERT_TRUE(manager.handleResponse(headers)); + ASSERT_TRUE(manager.isTokenBlocked("tenant-token-1")); + + nowMs += 9999; + EXPECT_TRUE(manager.isTokenBlocked("tenant-token-1")); + + nowMs += 1; + EXPECT_FALSE(manager.isTokenBlocked("tenant-token-1")); + EXPECT_FALSE(manager.isActive()); +} + TEST(KillSwitchManagerTests, handleResponse_NonNumericKillDuration_DoesNotThrowAndDoesNotBlock) { KillSwitchManager manager; diff --git a/tests/unittests/OfflineStorageTests_SQLite.cpp b/tests/unittests/OfflineStorageTests_SQLite.cpp index d5aa6808a..015e197d7 100644 --- a/tests/unittests/OfflineStorageTests_SQLite.cpp +++ b/tests/unittests/OfflineStorageTests_SQLite.cpp @@ -312,32 +312,31 @@ TEST_F(OfflineStorageTests_SQLite, ReservedRecordsAreReleasedAfterTimeout) ASSERT_THAT(offlineStorage->StoreRecord({"guid1", "token", EventLatency_Normal, EventPersistence_Normal, 1, {}}), true); ASSERT_THAT(offlineStorage->StoreRecord({"guid2", "token", EventLatency_Normal, EventPersistence_Normal, 1, {}}), true); TestRecordConsumer consumer; - // Reserve first for 2 secs - EXPECT_THAT(offlineStorage->GetAndReserveRecords(consumer, 2000, EventLatency_Unspecified, 1), true); + EXPECT_THAT(offlineStorage->GetAndReserveRecords(consumer, 5000, EventLatency_Unspecified, 1), true); ASSERT_THAT(consumer.records.size(), 1); consumer.records.clear(); - PAL::sleep(500); - - // Reserve second for 1 sec, first still unavailable - EXPECT_THAT(offlineStorage->GetAndReserveRecords(consumer, 1000, EventLatency_Unspecified, 1), true); + // The first record remains reserved, so the second call returns the other record. + EXPECT_THAT(offlineStorage->GetAndReserveRecords(consumer, 5000, EventLatency_Unspecified, 1), true); ASSERT_THAT(consumer.records.size(), 1); consumer.records.clear(); auto records = offlineStorage->GetRecords(true, EventLatency_Unspecified, 0); ASSERT_THAT(records.size(), 2); - int64_t waitUntilMs = 0; for (auto const& record : records) { - waitUntilMs = std::max(waitUntilMs, record.reservedUntil); + EXPECT_GT(record.reservedUntil, 1); } - while (PAL::getUtcSystemTimeMs() <= waitUntilMs + 250) + // Simulate lease expiry without depending on wall-clock sleeps or CI scheduling. + offlineStorage->Execute("UPDATE events SET reserved_until=1"); + records = offlineStorage->GetRecords(true, EventLatency_Unspecified, 0); + ASSERT_THAT(records.size(), 2); + for (auto const& record : records) { - PAL::sleep(50); + EXPECT_EQ(record.reservedUntil, 1); } - // Both records are timed out EXPECT_THAT(offlineStorage->GetAndReserveRecords(consumer, 1000), true); ASSERT_THAT(consumer.records.size(), 2); EXPECT_THAT(consumer.records[0].retryCount, 1);