diff --git a/lib/api/LogConfiguration.cpp b/lib/api/LogConfiguration.cpp index 23a7e53cd..0eb6581b2 100644 --- a/lib/api/LogConfiguration.cpp +++ b/lib/api/LogConfiguration.cpp @@ -19,6 +19,7 @@ namespace MAT_NS_BEGIN { { CFG_BOOL_ENABLE_ANALYTICS, false }, { CFG_INT_CACHE_FILE_SIZE, 3145728 }, { CFG_INT_RAM_QUEUE_SIZE, 524288 }, + { CFG_BOOL_ENABLE_BATCHED_STORAGE_FLUSH, true }, { CFG_BOOL_ENABLE_MULTITENANT, true }, { CFG_BOOL_ENABLE_DB_DROP_IF_FULL, false }, { CFG_INT_MAX_TEARDOWN_TIME, 0 }, @@ -51,6 +52,7 @@ namespace MAT_NS_BEGIN { { CFG_BOOL_ENABLE_ANALYTICS, src.enableLifecycleSession }, { CFG_INT_CACHE_FILE_SIZE, src.cacheFileSizeLimitInBytes }, { CFG_INT_RAM_QUEUE_SIZE, src.cacheMemorySizeLimitInBytes }, + { CFG_BOOL_ENABLE_BATCHED_STORAGE_FLUSH, true }, { CFG_BOOL_ENABLE_MULTITENANT, src.multiTenantEnabled }, { CFG_INT_MAX_TEARDOWN_TIME, src.maxTeardownUploadTimeInSec }, { CFG_INT_MAX_PENDING_REQ, src.maxPendingHTTPRequests }, @@ -128,4 +130,3 @@ namespace MAT_NS_BEGIN { } } MAT_NS_END - diff --git a/lib/api/LogManagerFactory.hpp b/lib/api/LogManagerFactory.hpp index 5e26267d8..63adfb646 100644 --- a/lib/api/LogManagerFactory.hpp +++ b/lib/api/LogManagerFactory.hpp @@ -67,7 +67,15 @@ namespace MAT_NS_BEGIN { // C++11 Magic Statics (N2660) static LogManagerFactory& instance() { - static LogManagerFactory impl; + // Deliberately never destroyed. LogManagerProvider::Release() must be + // able to walk this factory's registries during process teardown, but + // a normal function-local static's destruction order relative to that + // teardown call is unspecified -- if this were destroyed first, + // Release() would walk already-freed std::map nodes (a downstream + // consumer observed this as EXC_BAD_ACCESS in release() at process + // exit). Leaking one small, fixed-size object avoids the ordering + // hazard entirely; the OS reclaims it when the process exits. + static LogManagerFactory& impl = *new LogManagerFactory(); return impl; } diff --git a/lib/api/LogManagerImpl.cpp b/lib/api/LogManagerImpl.cpp index 24215c0cd..a06bb820b 100644 --- a/lib/api/LogManagerImpl.cpp +++ b/lib/api/LogManagerImpl.cpp @@ -7,6 +7,7 @@ #pragma warning(disable : 4459) #endif #include "LogManagerImpl.hpp" +#include #include "mat/config.h" #include "offline/LogSessionDataProvider.hpp" @@ -368,9 +369,27 @@ namespace MAT_NS_BEGIN LogManagerImpl::~LogManagerImpl() noexcept { - FlushAndTeardown(); - LOCKGUARD(ILogManagerInternal::managers_lock); - ILogManagerInternal::managers.erase(this); + try + { + FlushAndTeardown(); + } + catch (const std::exception& e) + { + std::fprintf(stderr, "Log manager teardown failed: %s\n", e.what()); + } + catch (...) + { + std::fputs("Log manager teardown failed with an unknown exception\n", stderr); + } + try + { + LOCKGUARD(ILogManagerInternal::managers_lock); + ILogManagerInternal::managers.erase(this); + } + catch (...) + { + std::fputs("Log manager registry cleanup failed\n", stderr); + } } size_t LogManagerImpl::GetDeadLoggerCount() @@ -959,19 +978,30 @@ namespace MAT_NS_BEGIN return true; } - void LogManagerImpl::EndActivity() + void LogManagerImpl::EndActivity() noexcept { - std::unique_lock lock(m_pause_mutex); - if (m_pause_active_count == 0) { - return; + try + { + std::unique_lock lock(m_pause_mutex); + if (m_pause_active_count == 0) { + return; + } + m_pause_active_count -= 1; + if (m_pause_active_count > 0) { + return; + } + if (m_pause_state == PauseState::Pausing) { + m_pause_state = PauseState::Paused; + m_pause_cv.notify_all(); + } } - m_pause_active_count -= 1; - if (m_pause_active_count > 0) { - return; + catch (const std::exception& e) + { + std::fprintf(stderr, "Failed to end telemetry activity: %s\n", e.what()); } - if (m_pause_state == PauseState::Pausing) { - m_pause_state = PauseState::Paused; - m_pause_cv.notify_all(); + catch (...) + { + std::fputs("Failed to end telemetry activity\n", stderr); } } } diff --git a/lib/api/LogManagerImpl.hpp b/lib/api/LogManagerImpl.hpp index 7dd7f7442..75e062868 100644 --- a/lib/api/LogManagerImpl.hpp +++ b/lib/api/LogManagerImpl.hpp @@ -306,7 +306,7 @@ namespace MAT_NS_BEGIN virtual void ResumeActivity() override; virtual void WaitPause() override; virtual bool StartActivity() override; - virtual void EndActivity() override; + virtual void EndActivity() noexcept override; protected: std::unique_ptr& GetSystem(); diff --git a/lib/api/Logger.cpp b/lib/api/Logger.cpp index aec4b9e52..75d3bd292 100644 --- a/lib/api/Logger.cpp +++ b/lib/api/Logger.cpp @@ -127,7 +127,8 @@ namespace MAT_NS_BEGIN Logger::~Logger() noexcept { - LOG_TRACE("%p: Destroyed", this); + // Intentionally empty — logging here triggers a static-destruction-order + // crash on iOS simulator (recursive_mutex used after teardown). } ISemanticContext* Logger::GetSemanticContext() const diff --git a/lib/config/RuntimeConfig_Default.hpp b/lib/config/RuntimeConfig_Default.hpp index 504aeefe3..4b2da9612 100644 --- a/lib/config/RuntimeConfig_Default.hpp +++ b/lib/config/RuntimeConfig_Default.hpp @@ -16,6 +16,7 @@ namespace MAT_NS_BEGIN {CFG_BOOL_ENABLE_ANALYTICS, false}, {CFG_INT_CACHE_FILE_SIZE, 3145728}, {CFG_INT_RAM_QUEUE_SIZE, 524288}, + {CFG_BOOL_ENABLE_BATCHED_STORAGE_FLUSH, true}, {CFG_BOOL_ENABLE_MULTITENANT, true}, {CFG_BOOL_ENABLE_DB_DROP_IF_FULL, false}, {CFG_INT_MAX_TEARDOWN_TIME, 1}, @@ -233,4 +234,3 @@ namespace MAT_NS_BEGIN } MAT_NS_END - diff --git a/lib/http/HttpClient_WinInet.cpp b/lib/http/HttpClient_WinInet.cpp index b1d3b4013..5a9d68a65 100644 --- a/lib/http/HttpClient_WinInet.cpp +++ b/lib/http/HttpClient_WinInet.cpp @@ -55,6 +55,9 @@ class WinInetRequestWrapper if (m_hWinInetRequest != nullptr) { ::InternetCloseHandle(m_hWinInetRequest); + } + if (m_hWinInetSession != nullptr) + { ::InternetCloseHandle(m_hWinInetSession); } } diff --git a/lib/http/HttpResponseDecoder.cpp b/lib/http/HttpResponseDecoder.cpp index 6014cec19..b46eeff98 100644 --- a/lib/http/HttpResponseDecoder.cpp +++ b/lib/http/HttpResponseDecoder.cpp @@ -67,13 +67,11 @@ namespace MAT_NS_BEGIN { break; case HttpResult_Aborted: - ctx->httpResponse = nullptr; outcome = Abort; break; case HttpResult_LocalFailure: case HttpResult_NetworkFailure: - ctx->httpResponse = nullptr; outcome = RetryNetwork; break; } @@ -132,7 +130,6 @@ namespace MAT_NS_BEGIN { evt.param2 = ctx->recordIdsAndTenantIds.size(); DispatchEvent(evt); } - ctx->httpResponse = nullptr; // eventsRejected(ctx); // FIXME: [MG] - investigate why ctx gets corrupt after eventsRejected requestAborted(ctx); break; diff --git a/lib/include/public/ILogConfiguration.hpp b/lib/include/public/ILogConfiguration.hpp index af1bc44c2..f1119c11d 100644 --- a/lib/include/public/ILogConfiguration.hpp +++ b/lib/include/public/ILogConfiguration.hpp @@ -154,6 +154,12 @@ namespace MAT_NS_BEGIN /// static constexpr const char* const CFG_INT_RAM_QUEUE_BUFFERS = "maxDBFlushQueues"; + /// + /// Batch records when flushing the RAM queue to disk storage. + /// Set to false to use per-record disk stores during flush. + /// + static constexpr const char* const CFG_BOOL_ENABLE_BATCHED_STORAGE_FLUSH = "enableBatchedStorageFlush"; + /// /// SQLite DB will be checkpointed when flushing. /// @@ -481,4 +487,3 @@ namespace MAT_NS_BEGIN } MAT_NS_END #endif - diff --git a/lib/include/public/ITaskDispatcher.hpp b/lib/include/public/ITaskDispatcher.hpp index 070f054bc..9fbeea9f1 100644 --- a/lib/include/public/ITaskDispatcher.hpp +++ b/lib/include/public/ITaskDispatcher.hpp @@ -121,6 +121,7 @@ namespace MAT_NS_BEGIN /// Amount of time to wait for if the task is currently executing /// True if successfully cancelled, else false virtual bool Cancel(Task* task, uint64_t waitTime = 0) = 0; + }; /// @endcond @@ -128,4 +129,3 @@ namespace MAT_NS_BEGIN } MAT_NS_END #endif // ITASKDISPATCHER_HPP - diff --git a/lib/jni/PrivacyGuard_jni.cpp b/lib/jni/PrivacyGuard_jni.cpp index 8fd23867a..5969ffc81 100644 --- a/lib/jni/PrivacyGuard_jni.cpp +++ b/lib/jni/PrivacyGuard_jni.cpp @@ -62,16 +62,22 @@ Java_com_microsoft_applications_events_PrivacyGuard_nativeInitializePrivacyGuard InitializationConfiguration config( reinterpret_cast(iLoggerNativePtr), CommonDataContext{}); + // InitializationConfiguration holds const char* pointers, so the backing + // std::string storage must outlive the PrivacyGuard construction below. + std::string notificationEventName, semanticContextEventName, summaryEventName; if (NotificationEventName != nullptr) { - config.NotificationEventName = JStringToStdString(env, NotificationEventName).c_str(); + notificationEventName = JStringToStdString(env, NotificationEventName); + config.NotificationEventName = notificationEventName.c_str(); } if (SemanticContextEventName != nullptr) { - config.SemanticContextNotificationEventName = JStringToStdString(env, SemanticContextEventName).c_str(); + semanticContextEventName = JStringToStdString(env, SemanticContextEventName); + config.SemanticContextNotificationEventName = semanticContextEventName.c_str(); } if (SummaryEventName != nullptr) { - config.SummaryEventName = JStringToStdString(env, SummaryEventName).c_str(); + summaryEventName = JStringToStdString(env, SummaryEventName); + config.SummaryEventName = summaryEventName.c_str(); } config.UseEventFieldPrefix = static_cast(UseEventFieldPrefix); @@ -119,16 +125,22 @@ Java_com_microsoft_applications_events_PrivacyGuard_nativeInitializePrivacyGuard machineIds, outOfScopeIdentifiers)); + // InitializationConfiguration holds const char* pointers, so the backing + // std::string storage must outlive the PrivacyGuard construction below. + std::string notificationEventName, semanticContextEventName, summaryEventName; if (NotificationEventName != NULL) { - config.NotificationEventName = JStringToStdString(env, NotificationEventName).c_str(); + notificationEventName = JStringToStdString(env, NotificationEventName); + config.NotificationEventName = notificationEventName.c_str(); } if (SemanticContextEventName != NULL) { - config.SemanticContextNotificationEventName = JStringToStdString(env, SemanticContextEventName).c_str(); + semanticContextEventName = JStringToStdString(env, SemanticContextEventName); + config.SemanticContextNotificationEventName = semanticContextEventName.c_str(); } if (SummaryEventName != NULL) { - config.SummaryEventName = JStringToStdString(env, SummaryEventName).c_str(); + summaryEventName = JStringToStdString(env, SummaryEventName); + config.SummaryEventName = summaryEventName.c_str(); } config.UseEventFieldPrefix = static_cast(UseEventFieldPrefix); diff --git a/lib/offline/MemoryStorage.cpp b/lib/offline/MemoryStorage.cpp index 1d4ec5664..77ff0fc7c 100644 --- a/lib/offline/MemoryStorage.cpp +++ b/lib/offline/MemoryStorage.cpp @@ -224,6 +224,16 @@ namespace MAT_NS_BEGIN { void MemoryStorage::DeleteRecords(const std::map & whereFilter) { + // An empty filter matches every record. Never silently wipe the whole + // in-memory queue from a no-op predicate; callers must use + // DeleteAllRecords() for an intentional full clear. This mirrors the + // fail-closed behavior of OfflineStorage_SQLite::DeleteRecords. + if (whereFilter.empty()) + { + LOG_WARN("DeleteRecords called with an empty filter; ignoring to avoid deleting all records."); + return; + } + auto matcher = [&](const StorageRecord &r, const std::map & whereFilter) { bool matched = true; diff --git a/lib/offline/OfflineStorageHandler.cpp b/lib/offline/OfflineStorageHandler.cpp index 52ce15515..c581089cd 100644 --- a/lib/offline/OfflineStorageHandler.cpp +++ b/lib/offline/OfflineStorageHandler.cpp @@ -7,9 +7,14 @@ #include "OfflineStorageFactory.hpp" #include "offline/MemoryStorage.hpp" +#include "offline/StorageRecordValidation.hpp" #include "ILogManager.hpp" +#include "utils/Utils.hpp" #include +#include +#include +#include #include #include @@ -50,6 +55,55 @@ namespace MAT_NS_BEGIN { } } + /// + /// RAII guard around ILogManager::StartActivity()/EndActivity(). Flush() + /// used to pair these manually (StartActivity() at the top, EndActivity() + /// on the last line), so an exception thrown by disk I/O or by + /// IOfflineStorageObserver::OnStorageRecordsSaved() partway through would + /// skip EndActivity() and permanently leak the pause-activity count -- + /// deadlocking every later FlushAndTeardown()'s WaitPause(). This guard + /// guarantees EndActivity() runs on every exit path, matching the existing + /// safe pattern used by PauseGuard (TransmissionPolicyManager.cpp) and + /// ActiveLoggerCall (Logger.cpp). + /// + class ActivityGuard + { + public: + explicit ActivityGuard(ILogManager& logManager) : + m_logManager(logManager), + m_active(logManager.StartActivity()) + { + } + + ~ActivityGuard() noexcept + { + if (m_active) + { + try + { + m_logManager.EndActivity(); + } + catch (const std::exception& e) + { + std::fprintf(stderr, "Failed to end telemetry activity: %s\n", e.what()); + } + catch (...) + { + std::fputs("Failed to end telemetry activity\n", stderr); + } + } + } + + ActivityGuard(ActivityGuard const&) = delete; + ActivityGuard& operator=(ActivityGuard const&) = delete; + + bool IsActive() const noexcept { return m_active; } + + private: + ILogManager& m_logManager; + bool m_active; + }; + bool OfflineStorageHandler::isKilled(StorageRecord const& record) { return ( @@ -64,7 +118,7 @@ namespace MAT_NS_BEGIN { if (!m_flushPending) return; } - LOG_INFO("Waiting for pending Flush (%p) to complete...", m_flushHandle.m_task); + LOG_INFO("Waiting for pending Flush (%p) to complete...", m_flushHandle.GetTask()); m_flushComplete.wait(); } @@ -163,65 +217,114 @@ namespace MAT_NS_BEGIN { void OfflineStorageHandler::Flush() { - if (!m_logManager.StartActivity()) { + ActivityGuard activityGuard(m_logManager); + if (!activityGuard.IsActive()) { + // The LogManager is shutting down, so the flush cannot run. Still + // signal completion and clear the pending flag so a concurrent + // WaitForFlush() (e.g. during teardown) does not block forever + // waiting for m_flushComplete. + LOCKGUARD(m_flushLock); + m_flushHandle.Cancel(); + m_flushComplete.post(); + m_flushPending = false; return; } - // Flush could be executed from context of worker thread, as well as from TPM and - // after HTTP callback. Make sure it is atomic / thread-safe. - LOCKGUARD(m_flushLock); - - // If item isn't scheduled yet, it gets canceled, so that we don't do two flushes. - // If we are running that item right now (our thread), then nothing happens other - // than the handle gets replaced by nullptr in this DeferredCallbackHandle obj. - m_flushHandle.Cancel(); - - size_t dbSizeBeforeFlush = (m_offlineStorageMemory != nullptr) ? m_offlineStorageMemory->GetSize() : 0; - if ((m_offlineStorageMemory) && (dbSizeBeforeFlush > 0) && (m_offlineStorageDisk)) + std::vector recordsToRecover; + try { - // This will block on and then take a lock for the duration of this move, and - // StoreRecord() will then block until the move completes. - auto records = m_offlineStorageMemory->GetRecords(false, EventLatency_Unspecified); - std::vector ids; - - // TODO: [MG] - consider running the batch in transaction - // if (sqlite) - // sqlite->Execute("BEGIN"); + // Flush could be executed from context of worker thread, as well as from TPM and + // after HTTP callback. Make sure it is atomic / thread-safe. + LOCKGUARD(m_flushLock); - size_t totalSaved = m_offlineStorageDisk->StoreRecords(records); + // If item isn't scheduled yet, it gets canceled, so that we don't do two flushes. + // If we are running that item right now (our thread), then nothing happens other + // than the handle reporting nullptr once that task finishes. + m_flushHandle.Cancel(); - // TODO: [MG] - consider running the batch in transaction - // if (sqlite) - // sqlite->Execute("END"); + size_t dbSizeBeforeFlush = (m_offlineStorageMemory != nullptr) ? m_offlineStorageMemory->GetSize() : 0; + if ((m_offlineStorageMemory) && (dbSizeBeforeFlush > 0) && (m_offlineStorageDisk)) + { + // Drain the in-memory queue into a local batch. Records are removed + // from memory here; any that fail to persist below are re-inserted, so + // a disk write failure does not silently lose events. Draining (rather + // than reserving) keeps only a single copy of each record in flight and + // avoids stamping a reservation lease that the Room backend would + // persist to disk. + recordsToRecover = m_offlineStorageMemory->GetRecords(false, EventLatency_Unspecified); + + size_t totalSaved = 0; + if (IsBatchedStorageFlushEnabled()) + { + // Persist the drained batch to disk in a single transaction. + // StoreRecords() commits as many records as it durably can and + // returns that count. Records it can never store (e.g. ones failing + // validation, reported separately) are dropped from the batch rather + // than counted, so a return of 0 with records still queued means a + // transient failure committed nothing -- return those records to the + // in-memory queue for retry. No events are lost, and a rolled-back + // batch leaves nothing on disk, so re-queuing cannot create duplicates + // (the events table has no unique record_id constraint). A non-zero + // count means those records are durably stored; do not re-queue. + totalSaved = m_offlineStorageDisk->StoreRecords(recordsToRecover); + if (totalSaved == 0 && !recordsToRecover.empty()) + { + LOG_WARN("Flush: disk store failed for the batch of %zu records; returning to the queue for retry", + recordsToRecover.size()); + ReturnRecordsToMemory(recordsToRecover); + } + } + else + { + totalSaved = StoreRecordsIndividually(recordsToRecover); + } - // Delete records from reserved on flush - HttpHeaders dummy; - bool fromMemory = true; - m_offlineStorageMemory->DeleteRecords(ids, dummy, fromMemory); + // Persistence and retry handling are complete; a later exception + // must not requeue records that were already committed. + recordsToRecover.clear(); - // Notify event listener about the records cached - OnStorageRecordsSaved(totalSaved); + if (m_offlineStorageMemory->GetSize() > dbSizeBeforeFlush) + { + // We managed to accumulate as much data as we had before the flush, + // means we cannot keep up flushing at the same speed as incoming + // obviously because the disk is slower than ram. + LOG_WARN("Data is arriving too fast!"); + } + OnStorageRecordsSaved(totalSaved); + } - if (m_offlineStorageMemory->GetSize() > dbSizeBeforeFlush) + // Checkpoint DB + if (m_offlineStorageDisk && m_config.HasConfig(CFG_BOOL_CHECKPOINT_DB_ON_FLUSH) && m_config[CFG_BOOL_CHECKPOINT_DB_ON_FLUSH]) { - // We managed to accumulate as much data as we had before the flush, - // means we cannot keep up flushing at the same speed as incoming - // obviously because the disk is slower than ram. - LOG_WARN("Data is arriving too fast!"); + m_offlineStorageDisk->Flush(); } - } - // Checkpoint DB - if (m_config.HasConfig(CFG_BOOL_CHECKPOINT_DB_ON_FLUSH) && m_config[CFG_BOOL_CHECKPOINT_DB_ON_FLUSH]) + m_isStorageFullNotificationSend = false; + m_flushComplete.post(); + m_flushPending = false; + } + catch (...) { - m_offlineStorageDisk->Flush(); + std::exception_ptr failure = std::current_exception(); + try + { + if (m_offlineStorageMemory && !recordsToRecover.empty()) + { + ReturnRecordsToMemory(recordsToRecover); + } + } + catch (const std::exception& e) + { + std::fprintf(stderr, "Failed to recover records after flush failure: %s\n", e.what()); + } + catch (...) + { + std::fputs("Failed to recover records after flush failure\n", stderr); + } + LOCKGUARD(m_flushLock); + m_flushComplete.post(); + m_flushPending = false; + std::rethrow_exception(failure); } - - m_isStorageFullNotificationSend = false; - - // Flush is done, notify the waiters - m_flushComplete.post(); - m_flushPending = false; - m_logManager.EndActivity(); } bool OfflineStorageHandler::StoreRecord(StorageRecord const& record) @@ -247,7 +350,21 @@ namespace MAT_NS_BEGIN { // are selected and removed from the cache (but will // not block for the subsequent handoff to persistent // storage) - m_offlineStorageMemory->StoreRecord(record); + if (!m_offlineStorageMemory->StoreRecord(record)) + { + if (record.latency == EventLatency_Off) + { + // MemoryStorage intentionally returns false for latency-off + // records to mean "drop without storing", not "storage + // failed". Keep the handler's false return reserved for + // genuine storage failures so StorageObserver does not + // misclassify this normal drop as a persistence error. + return true; + } + LOG_ERROR("Failed to store event %s:%s in memory queue", + tenantTokenToId(record.tenantToken).c_str(), record.id.c_str()); + return false; + } } // Perform periodic flush to disk @@ -260,7 +377,7 @@ namespace MAT_NS_BEGIN { m_flushPending = true; m_flushComplete.Reset(); m_flushHandle = PAL::scheduleTask(&m_taskDispatcher, 0, this, &OfflineStorageHandler::Flush); - LOG_INFO("Requested Flush (%p)", m_flushHandle.m_task); + LOG_INFO("Requested Flush (%p)", m_flushHandle.GetTask()); } m_flushLock.unlock(); } @@ -272,7 +389,9 @@ namespace MAT_NS_BEGIN { { if (record.persistence != EventPersistence::EventPersistence_DoNotStoreOnDisk) { - m_offlineStorageDisk->StoreRecord(record); + // Propagate a synchronous disk write failure to the caller so a + // failed store is not counted as successfully persisted. + return m_offlineStorageDisk->StoreRecord(record); } } } @@ -280,6 +399,109 @@ namespace MAT_NS_BEGIN { return true; } + bool OfflineStorageHandler::IsBatchedStorageFlushEnabled() + { + return !m_config.HasConfig(CFG_BOOL_ENABLE_BATCHED_STORAGE_FLUSH) || + m_config[CFG_BOOL_ENABLE_BATCHED_STORAGE_FLUSH]; + } + + void OfflineStorageHandler::ReportInvalidDiskRecord(StorageRecord const& record) + { + LOG_ERROR("Flush: dropping event %s:%s: Invalid parameters", + tenantTokenToId(record.tenantToken).c_str(), record.id.c_str()); + OnStorageFailed("Invalid parameters"); + } + + size_t OfflineStorageHandler::StoreRecordsIndividually(std::vector const& records) + { + size_t totalSaved = 0; + std::vector recordsToRetry; + + for (auto it = records.begin(); it != records.end(); ++it) + { + if (!IsValidDiskStorageRecord(*it)) + { + ReportInvalidDiskRecord(*it); + continue; + } + + if (m_offlineStorageDisk->StoreRecord(*it)) + { + ++totalSaved; + continue; + } + + for (auto retryIt = it; retryIt != records.end(); ++retryIt) + { + if (IsValidDiskStorageRecord(*retryIt)) + { + recordsToRetry.push_back(*retryIt); + } + else + { + ReportInvalidDiskRecord(*retryIt); + } + } + break; + } + + if (!recordsToRetry.empty()) + { + LOG_WARN("Flush: per-record disk store failed after saving %zu of %zu records; returning %zu records to the queue for retry", + totalSaved, records.size(), recordsToRetry.size()); + ReturnRecordsToMemory(recordsToRetry); + } + + return totalSaved; + } + + size_t OfflineStorageHandler::ReturnRecordsToMemory(std::vector const& records) + { + size_t returned = 0; + DroppedMap dropped; + + for (auto const& record : records) + { + try + { + if (m_offlineStorageMemory && m_offlineStorageMemory->StoreRecord(record)) + { + ++returned; + continue; + } + LOG_ERROR("Flush: failed to return event %s:%s to memory queue after disk store failure; dropping record", + tenantTokenToId(record.tenantToken).c_str(), record.id.c_str()); + dropped[record.tenantToken]++; + } + catch (const std::exception& e) + { + std::fprintf(stderr, "Failed to recover a record after flush failure: %s\n", e.what()); + } + catch (...) + { + std::fputs("Failed to recover a record after flush failure\n", stderr); + } + } + + if (!dropped.empty()) + { + try + { + OnStorageRecordsDropped(dropped); + } + catch (const std::exception& e) + { + std::fprintf(stderr, "Failed to report dropped records after flush failure: %s\n", e.what()); + } + catch (...) + { + std::fputs("Failed to report dropped records after flush failure\n", stderr); + } + } + + return returned; + } + size_t OfflineStorageHandler::StoreRecords(std::vector& records) { size_t stored = 0; diff --git a/lib/offline/OfflineStorageHandler.hpp b/lib/offline/OfflineStorageHandler.hpp index 9a1131aff..21d5702e0 100644 --- a/lib/offline/OfflineStorageHandler.hpp +++ b/lib/offline/OfflineStorageHandler.hpp @@ -25,6 +25,8 @@ namespace MAT_NS_BEGIN { class OfflineStorageHandler final : public IOfflineStorage, public IOfflineStorageObserver { + friend class OfflineStorageHandlerTestPeer; + public: OfflineStorageHandler(ILogManager& logManager, IRuntimeConfig& runtimeConfig, ITaskDispatcher& taskDispatcher); virtual ~OfflineStorageHandler() override; @@ -85,8 +87,8 @@ namespace MAT_NS_BEGIN { std::unique_ptr m_offlineStorageMemory; std::shared_ptr m_offlineStorageDisk; - bool m_readFromMemory; - unsigned m_lastReadCount; + std::atomic m_readFromMemory; + std::atomic m_lastReadCount; bool m_shutdownStarted; unsigned m_memoryDbSize; @@ -100,6 +102,10 @@ namespace MAT_NS_BEGIN { private: void WaitForFlush(); + bool IsBatchedStorageFlushEnabled(); + void ReportInvalidDiskRecord(StorageRecord const& record); + size_t StoreRecordsIndividually(std::vector const& records); + size_t ReturnRecordsToMemory(std::vector const& records); }; diff --git a/lib/offline/OfflineStorage_SQLite.cpp b/lib/offline/OfflineStorage_SQLite.cpp index b9b2ed83d..cf3cb8ac3 100644 --- a/lib/offline/OfflineStorage_SQLite.cpp +++ b/lib/offline/OfflineStorage_SQLite.cpp @@ -8,6 +8,7 @@ #include "OfflineStorage_SQLite.hpp" #include "ILogManager.hpp" #include "SQLiteWrapper.hpp" +#include "StorageRecordValidation.hpp" #include "utils/StringUtils.hpp" #include #include @@ -23,6 +24,8 @@ namespace MAT_NS_BEGIN { class DbTransaction { SqliteDB* m_db; + bool m_rollback = false; + bool m_finished = false; public: bool locked; @@ -34,11 +37,43 @@ namespace MAT_NS_BEGIN { } } + // Discard the transaction (ROLLBACK) instead of committing it on destruction. + void markForRollback() + { + m_rollback = true; + } + + // Commit the transaction now and report whether COMMIT succeeded. On a + // COMMIT failure the transaction is rolled back so it is never left open, + // and false is returned so the caller does not treat undurable writes as + // stored. After this call the destructor performs no further COMMIT/ROLLBACK. + bool commit() + { + if (!locked || m_finished) + { + return false; + } + m_finished = true; + if (m_db->unlock()) + { + return true; + } + m_db->rollback(); + return false; + } + ~DbTransaction() { - if (locked) + if (locked && !m_finished) { - m_db->unlock(); + if (m_rollback) + { + m_db->rollback(); + } + else + { + m_db->unlock(); + } } } }; @@ -147,40 +182,31 @@ namespace MAT_NS_BEGIN { m_db->execute(command.c_str()); } - bool OfflineStorage_SQLite::StoreRecord(StorageRecord const& record) + bool OfflineStorage_SQLite::isValidRecord(StorageRecord const& record) const { - // TODO: [MG] - this works, but may not play nicely with several LogManager instances - // static SqliteStatement sql_insert(*m_db, m_stmtInsertEvent_id_tenant_prio_ts_data); - - if (record.id.empty() || record.tenantToken.empty() || static_cast(record.latency) < 0 || record.timestamp <= 0) { + if (!IsValidDiskStorageRecord(record)) { LOG_ERROR("Failed to store event %s:%s: Invalid parameters", tenantTokenToId(record.tenantToken).c_str(), record.id.c_str()); m_observer->OnStorageFailed("Invalid parameters"); return false; } + return true; + } - if (!m_db) { - LOG_ERROR("Failed to store event %s:%s: Database is not open", + bool OfflineStorage_SQLite::insertRecordUnsafe(StorageRecord const& record) + { + if (!SqliteStatement(*m_db, m_stmtInsertEvent_id_tenant_prio_ts_data).execute(record.id, record.tenantToken, static_cast(record.latency), static_cast(record.persistence), record.timestamp, record.blob)) + { + LOG_ERROR("Failed to store event %s:%s: database write failed", tenantTokenToId(record.tenantToken).c_str(), record.id.c_str()); - m_observer->OnStorageOpenFailed("Database is not open"); return false; } + m_DbSizeEstimate += record.id.size() + record.tenantToken.size() + record.blob.size(); + return true; + } - { -#ifdef ENABLE_LOCKING - LOCKGUARD(m_lock); - DbTransaction transaction(m_db.get()); - if (!transaction.locked) - { - LOG_ERROR("Failed to store event %s:%s: Database error", tenantTokenToId(record.tenantToken).c_str(), record.id.c_str()); - m_observer->OnStorageFailed("Database error"); - return false; - } -#endif - SqliteStatement(*m_db, m_stmtInsertEvent_id_tenant_prio_ts_data).execute(record.id, record.tenantToken, static_cast(record.latency), static_cast(record.persistence), record.timestamp, record.blob); - m_DbSizeEstimate += record.id.size() + record.tenantToken.size() + record.blob.size(); - } - + void OfflineStorage_SQLite::checkStorageSizeLimits() + { if ((m_DbSizeNotificationLimit != 0) && (m_DbSizeEstimate>m_DbSizeNotificationLimit)) { auto now = PAL::getMonotonicTimeMs(); @@ -210,20 +236,159 @@ namespace MAT_NS_BEGIN { m_resizing = false; } } + } - return true; + bool OfflineStorage_SQLite::StoreRecord(StorageRecord const& record) + { + // TODO: [MG] - this works, but may not play nicely with several LogManager instances + // static SqliteStatement sql_insert(*m_db, m_stmtInsertEvent_id_tenant_prio_ts_data); + + if (!isValidRecord(record)) { + return false; + } + + if (!m_db) { + LOG_ERROR("Failed to store event %s:%s: Database is not open", + tenantTokenToId(record.tenantToken).c_str(), record.id.c_str()); + m_observer->OnStorageOpenFailed("Database is not open"); + return false; + } + + bool stored = false; + { +#ifdef ENABLE_LOCKING + LOCKGUARD(m_lock); + DbTransaction transaction(m_db.get()); + if (!transaction.locked) + { + LOG_ERROR("Failed to store event %s:%s: Database error", tenantTokenToId(record.tenantToken).c_str(), record.id.c_str()); + m_observer->OnStorageFailed("Database error"); + return false; + } + if (insertRecordUnsafe(record)) + { + // Verify the COMMIT: a COMMIT that fails must not be reported as a + // successful store, or the caller treats an undurable write as saved. + stored = transaction.commit(); + if (!stored) + { + m_DbSizeEstimate -= std::min(m_DbSizeEstimate.load(), + record.id.size() + record.tenantToken.size() + record.blob.size()); + } + } + else + { + transaction.markForRollback(); + } +#else + stored = insertRecordUnsafe(record); +#endif + } + + if (!stored) { + // Report the write failure after the transaction has closed, so the + // observer callback never runs while BEGIN EXCLUSIVE is held. + m_observer->OnStorageFailed("Database write failed"); + } + + // Run the size-limit check after the transaction, matching the original + // per-record path (which ran it on every StoreRecord call). + checkStorageSizeLimits(); + + return stored; } size_t OfflineStorage_SQLite::StoreRecords(std::vector & records) { - size_t stored = 0; - for (auto & i : records) { - if (StoreRecord(i)) { - ++stored; + if (records.empty()) { + return 0; + } + + // Drop invalid records up front (each is reported by isValidRecord) so a + // permanently-invalid record is discarded rather than failing the whole + // batch. Removing them from the vector means a caller that re-queues on a + // short return (e.g. Flush) never re-queues a poison record -- which would + // be re-drained and re-rejected on every flush, blocking every valid record + // behind it -- while the valid remainder stays all-or-nothing. + records.erase( + std::remove_if(records.begin(), records.end(), + [this](StorageRecord const& record) { return !isValidRecord(record); }), + records.end()); + + if (records.empty()) { + // Every record was invalid (already reported). + return 0; + } + + if (!m_db) { + LOG_ERROR("Failed to store %zu events: Database is not open", records.size()); + m_observer->OnStorageOpenFailed("Database is not open"); + return 0; + } + + size_t addedSize = 0; + bool committed = false; + { + // Batch all inserts into a single transaction: one BEGIN EXCLUSIVE / + // COMMIT (one fsync) for the whole flush instead of one per record. + // All-or-nothing: if any insert OR the COMMIT fails the transaction is + // rolled back, so callers (e.g. Flush) can re-queue the whole batch + // without risking duplicate rows (the events table has no unique + // record_id constraint). + bool allInserted = true; +#ifdef ENABLE_LOCKING + LOCKGUARD(m_lock); + DbTransaction transaction(m_db.get()); + if (!transaction.locked) + { + LOG_ERROR("Failed to store %zu events: Database error", records.size()); + m_observer->OnStorageFailed("Database error"); + return 0; + } +#endif + for (auto const& r : records) { + if (insertRecordUnsafe(r)) { + addedSize += r.id.size() + r.tenantToken.size() + r.blob.size(); + } + else { + allInserted = false; + break; + } + } + +#ifdef ENABLE_LOCKING + if (allInserted) { + // Verify the COMMIT: a COMMIT that fails (e.g. SQLITE_FULL/IOERR) + // must not be reported as success, or Flush would drop the records + // it already drained from memory. + committed = transaction.commit(); + } + else { + transaction.markForRollback(); + } +#else + committed = allInserted; +#endif + + if (!committed) { + // Nothing durably stored; undo the size estimate added by the + // (rolled-back) inserts. + m_DbSizeEstimate -= std::min(m_DbSizeEstimate.load(), addedSize); } } - return stored; + + if (!committed) { + // The whole batch was rolled back after an insert or COMMIT failure; + // report once. + m_observer->OnStorageFailed("Database write failed"); + } + + // Run the size-full notification / resize check once after the batch, + // matching the original per-record path (which ran it on every insert). + checkStorageSizeLimits(); + + return committed ? records.size() : 0; } // Debug routine to print record count in the DB @@ -1064,4 +1229,3 @@ namespace MAT_NS_BEGIN { } MAT_NS_END #endif - diff --git a/lib/offline/OfflineStorage_SQLite.hpp b/lib/offline/OfflineStorage_SQLite.hpp index 18643cde5..1d32a4c77 100644 --- a/lib/offline/OfflineStorage_SQLite.hpp +++ b/lib/offline/OfflineStorage_SQLite.hpp @@ -122,6 +122,15 @@ namespace MAT_NS_BEGIN { private: size_t GetRecordCountUnsafe(EventLatency latency) const; + + // Validate a record's required fields; reports OnStorageFailed on rejection. + bool isValidRecord(StorageRecord const& record) const; + // Insert one already-validated record. Caller must hold m_lock and have an + // active DbTransaction (when ENABLE_LOCKING). Updates m_DbSizeEstimate. + // Returns false (without updating the size estimate) if the insert fails. + bool insertRecordUnsafe(StorageRecord const& record); + // Run the DB-size-full notification and resize checks (after inserts). + void checkStorageSizeLimits(); }; diff --git a/lib/offline/SQLiteWrapper.hpp b/lib/offline/SQLiteWrapper.hpp index 2a5f0d108..84f33f115 100644 --- a/lib/offline/SQLiteWrapper.hpp +++ b/lib/offline/SQLiteWrapper.hpp @@ -224,6 +224,16 @@ namespace MAT_NS_BEGIN { { } + ~SqliteDB() + { + // Finalize prepared statements and close the database even if + // shutdown() was not called explicitly (e.g. the owning storage was + // destroyed without Shutdown()). shutdown() is idempotent -- it + // returns immediately once m_db is null -- so an earlier explicit + // shutdown() makes this a no-op. + shutdown(); + } + bool initialize(std::string const& filename, bool deletePrevious, size_t maxHeapLimit = 0) { int result = SQLITE_OK; @@ -490,6 +500,13 @@ namespace MAT_NS_BEGIN { return isOK(sqlite3_exec("COMMIT;")); } + /** + * @brief Roll back (discard) the current DB transaction. + */ + bool rollback() { + return isOK(sqlite3_exec("ROLLBACK;")); + } + bool lock() { #ifndef NDEBUG unsigned count = 0; diff --git a/lib/offline/StorageRecordValidation.hpp b/lib/offline/StorageRecordValidation.hpp new file mode 100644 index 000000000..23447a11f --- /dev/null +++ b/lib/offline/StorageRecordValidation.hpp @@ -0,0 +1,21 @@ +// +// Copyright (c) Microsoft Corporation. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// + +#ifndef STORAGERECORDVALIDATION_HPP +#define STORAGERECORDVALIDATION_HPP + +#include "IOfflineStorage.hpp" + +namespace MAT_NS_BEGIN { + + inline bool IsValidDiskStorageRecord(StorageRecord const& record) + { + return !(record.id.empty() || record.tenantToken.empty() || + static_cast(record.latency) < 0 || record.timestamp <= 0); + } + +} MAT_NS_END + +#endif diff --git a/lib/pal/PAL.cpp b/lib/pal/PAL.cpp index 0fc28abfb..300a98e28 100644 --- a/lib/pal/PAL.cpp +++ b/lib/pal/PAL.cpp @@ -60,7 +60,18 @@ namespace PAL_NS_BEGIN { PlatformAbstractionLayer& GetPAL() noexcept { - static PlatformAbstractionLayer pal; + // Deliberately never destroyed. PAL::shutdown() (called from + // LogManagerImpl::FlushAndTeardown()) must find this object's members + // still alive, but PAL is constructed lazily on first use, so whether + // this function-local static is destroyed before or after that + // teardown call depends on runtime timing, not source order -- if it + // is destroyed first, shutdown() releases shared_ptr members of an + // already-destroyed object (a downstream consumer observed this as + // intermittent EXC_BAD_ACCESS in ~shared_ptr at + // process exit). Leaking one fixed-size object avoids the ordering + // hazard entirely: shutdown() already performs the real resource + // teardown explicitly, and the OS reclaims the object at process exit. + static PlatformAbstractionLayer& pal = *new PlatformAbstractionLayer(); return pal; } diff --git a/lib/pal/TaskDispatcher.hpp b/lib/pal/TaskDispatcher.hpp index ec6f2f690..4608a6c59 100644 --- a/lib/pal/TaskDispatcher.hpp +++ b/lib/pal/TaskDispatcher.hpp @@ -16,6 +16,7 @@ #include #include #include +#include #include #include "ITaskDispatcher.hpp" @@ -25,6 +26,15 @@ namespace PAL_NS_BEGIN { namespace detail { + struct TaskLifetimeState + { + TaskLifetimeState() : + task(nullptr) + {} + + std::atomic task; + }; + template class TaskCall : public Task { @@ -48,14 +58,35 @@ namespace PAL_NS_BEGIN { this->TargetTime = targetTime; } + TaskCall(TCall& call, int64_t targetTime, std::shared_ptr lifetimeState) : + Task(), + m_call(call), + m_lifetimeState(std::move(lifetimeState)) + { + this->TypeName = TYPENAME(call); + this->Type = Task::TimedCall; + this->TargetTime = targetTime; + if (m_lifetimeState) { + m_lifetimeState->task.store(this, std::memory_order_release); + } + } + virtual void operator()() override { m_call(); } - virtual ~TaskCall() noexcept = default; + virtual ~TaskCall() noexcept + { + if (m_lifetimeState) { + m_lifetimeState->task.store(nullptr, std::memory_order_release); + } + } const TCall m_call; + + private: + std::shared_ptr m_lifetimeState; }; } // namespace detail @@ -63,14 +94,11 @@ namespace PAL_NS_BEGIN { class DeferredCallbackHandle { public: - std::mutex m_mutex; - MAT::Task* m_task = nullptr; - MAT::ITaskDispatcher* m_taskDispatcher = nullptr; - - DeferredCallbackHandle(MAT::Task* task, MAT::ITaskDispatcher* taskDispatcher) : - m_task(task), + DeferredCallbackHandle(std::shared_ptr taskLifetimeState, MAT::ITaskDispatcher* taskDispatcher) : + m_taskLifetimeState(std::move(taskLifetimeState)), m_taskDispatcher(taskDispatcher) { } - DeferredCallbackHandle() {} + + DeferredCallbackHandle() = default; DeferredCallbackHandle(DeferredCallbackHandle&& h) { *this = std::move(h); @@ -78,28 +106,44 @@ namespace PAL_NS_BEGIN { DeferredCallbackHandle& operator=(DeferredCallbackHandle&& other) { + if (this == &other) { + return *this; + } + std::lock_guard lock(m_mutex); std::lock_guard otherLock(other.m_mutex); - m_task = other.m_task; - other.m_task = nullptr; + m_taskLifetimeState = std::move(other.m_taskLifetimeState); m_taskDispatcher = other.m_taskDispatcher; + other.m_taskDispatcher = nullptr; return *this; } + MAT::Task* GetTask() const + { + std::lock_guard lock(m_mutex); + return (m_taskLifetimeState != nullptr) ? m_taskLifetimeState->task.load(std::memory_order_acquire) : nullptr; + } + bool Cancel(uint64_t waitTime = 0) { std::lock_guard lock(m_mutex); - if (m_task) + MAT::Task* task = (m_taskLifetimeState != nullptr) ? m_taskLifetimeState->task.load(std::memory_order_acquire) : nullptr; + if (task) { - bool result = (m_taskDispatcher != nullptr) && (m_taskDispatcher->Cancel(m_task, waitTime)); - return result; + bool result = (m_taskDispatcher != nullptr) && (m_taskDispatcher->Cancel(task, waitTime)); + return result || ((m_taskLifetimeState != nullptr) && (m_taskLifetimeState->task.load(std::memory_order_acquire) == nullptr)); } else { // Canceled nothing successfully return true; } } + + private: + mutable std::mutex m_mutex; + std::shared_ptr m_taskLifetimeState; + MAT::ITaskDispatcher* m_taskDispatcher = nullptr; }; template @@ -121,9 +165,18 @@ namespace PAL_NS_BEGIN { DeferredCallbackHandle scheduleTask(MAT::ITaskDispatcher* taskDispatcher, unsigned delayMs, TObject* obj, void (TObject::*func)(TFuncArgs...), TPassedArgs&&... args) { auto bound = std::bind(std::mem_fn(func), obj, std::forward(args)...); - auto task = new detail::TaskCall(bound, getMonotonicTimeMs() + (int64_t)delayMs); + auto taskLifetime = std::make_shared(); + auto task = new detail::TaskCall(bound, getMonotonicTimeMs() + (int64_t)delayMs, taskLifetime); taskDispatcher->Queue(task); - return DeferredCallbackHandle(task, taskDispatcher); + // Queue() is void; an SDK dispatcher that rejects by deleting the task + // synchronously clears this state before Queue() returns, and the task + // destructor also clears it after normal asynchronous completion so a + // later Cancel() never touches a stale Task*. + if (taskLifetime->task.load(std::memory_order_acquire) == nullptr) + { + return DeferredCallbackHandle(); + } + return DeferredCallbackHandle(taskLifetime, taskDispatcher); } template @@ -135,4 +188,3 @@ namespace PAL_NS_BEGIN { } PAL_NS_END #endif - diff --git a/lib/pal/WorkerThread.cpp b/lib/pal/WorkerThread.cpp index 3adfb9e61..bdba6eec9 100644 --- a/lib/pal/WorkerThread.cpp +++ b/lib/pal/WorkerThread.cpp @@ -7,6 +7,8 @@ #include "pal/PAL.hpp" #include +#include +#include #if defined(MATSDK_PAL_CPP11) || defined(MATSDK_PAL_WIN32) @@ -29,6 +31,14 @@ namespace PAL_NS_BEGIN { { protected: std::thread m_hThread; + // The worker thread's own id, captured under m_lock once threadFunc starts. + // onLastReferenceReleased() reads it (under m_lock) rather than m_hThread.get_id() + // to detect "am I running on my own worker thread?", because m_hThread.get_id() + // returns the default not-a-thread id after a detach() -- so this keeps + // self-dispose detection correct even if the thread was detached first. A plain + // std::thread::id guarded by m_lock is used rather than std::atomic, + // which is not portable (std::thread::id is not guaranteed trivially copyable). + std::thread::id m_workerId; std::recursive_mutex m_lock; std::timed_mutex m_execution_mutex; @@ -36,16 +46,21 @@ namespace PAL_NS_BEGIN { std::list m_queue; std::list m_timerQueue; Event m_event; - MAT::Task* m_itemInProgress; - int count = 0; + std::atomic m_itemInProgress; + bool m_shuttingDown = false; + std::mutex m_joinLock; + // Set when the last reference is released by a task running on this worker + // thread, so threadFunc performs the final delete after its loop breaks + // (see onLastReferenceReleased() and WorkerThreadFactory::Create()). + std::atomic m_disposeFromThread { false }; public: WorkerThread() { - m_itemInProgress = nullptr; + m_itemInProgress.store(nullptr, std::memory_order_relaxed); m_hThread = std::thread(WorkerThread::threadFunc, static_cast(this)); - LOG_INFO("Started new thread %u", m_hThread.get_id()); + LOG_INFO("Started new thread %zu", std::hash{}(m_hThread.get_id())); } ~WorkerThread() @@ -53,34 +68,116 @@ namespace PAL_NS_BEGIN { Join(); } + private: + void enqueueShutdownItemLocked() + { + if (!m_shuttingDown) { + m_shuttingDown = true; + m_queue.push_back(new WorkerThreadShutdownItem()); + m_event.post(); + } + } + + void drainPendingTasksLocked() + { + for (auto task : m_queue) { delete task; } + m_queue.clear(); + for (auto task : m_timerQueue) { delete task; } + m_timerQueue.clear(); + } + + public: void Join() final { - auto item = new WorkerThreadShutdownItem(); - Queue(item); + LOCKGUARD(m_joinLock); std::thread::id this_id = std::this_thread::get_id(); - try { - if (m_hThread.joinable() && (m_hThread.get_id() != this_id)) - m_hThread.join(); - else + std::thread threadToJoin; + bool joined = false; + { + LOCKGUARD(m_lock); + enqueueShutdownItemLocked(); + if (!m_hThread.joinable()) { + return; + } + if (m_hThread.get_id() == this_id) { m_hThread.detach(); + } else { + threadToJoin = std::move(m_hThread); + } + } + try { + if (threadToJoin.joinable()) { + threadToJoin.join(); + joined = true; + } + } + catch (const std::system_error& e) { + LOG_ERROR("Thread join/detach failed: [%d] %s", e.code().value(), e.what()); + std::terminate(); + } + catch (const std::exception& e) { + LOG_ERROR("Thread join/detach failed: %s", e.what()); + std::terminate(); } - catch (...) {}; - // TODO: [MG] - investigate if we ever drop work items on shutdown. - if (!m_queue.empty()) - { - LOG_WARN("m_queue is not empty!"); + // Log pending work in both paths so operators can see if + // shutdown is dropping tasks. + LOCKGUARD(m_lock); + if (!m_queue.empty()) { + LOG_WARN("Shutdown with %zu queued task(s) pending", m_queue.size()); } - if (!m_timerQueue.empty()) + if (!m_timerQueue.empty()) { + LOG_WARN("Shutdown with %zu timer(s) pending", m_timerQueue.size()); + } + + // Clean up any tasks remaining in the queues after shutdown. + // Only safe after join() — the thread has fully exited. + // After detach(), the thread still needs the shutdown item + // and may still be accessing the queues. + if (joined) { + drainPendingTasksLocked(); + } + } + + // Invoked by the shared_ptr deleter when the last reference is released. + // Returns true if the caller should delete the object, false if deletion was + // deferred to the worker thread. The worker is shared process-wide, so the + // last reference can be dropped by a task running on the worker thread itself + // (e.g. a task that tears down its LogManager/PAL). In that case threadFunc is + // still on the stack below the task and keeps touching members after the task + // returns, so freeing the object here would be a use-after-free: instead + // detach, signal shutdown, mark the thread to delete itself once its loop + // breaks, and leave the object alive. On any other thread it is safe to delete + // immediately (~WorkerThread joins the worker first). + bool onLastReferenceReleased() + { + LOCKGUARD(m_lock); + if (m_workerId == std::this_thread::get_id()) { - LOG_WARN("m_timerQueue is not empty!"); + enqueueShutdownItemLocked(); + m_disposeFromThread.store(true, std::memory_order_release); + try { + if (m_hThread.joinable()) { + m_hThread.detach(); + } + } + catch (const std::exception& e) { + LOG_ERROR("Worker self-detach failed: %s", e.what()); + } + return false; } + return true; } void Queue(MAT::Task* item) final { - LOG_INFO("queue item=%p", &item); + LOG_INFO("queue item=%p", static_cast(item)); LOCKGUARD(m_lock); + if (m_shuttingDown) { + LOG_WARN("Dropping queued task %p during shutdown", static_cast(item)); + delete item; + return; + } if (item->Type == MAT::Task::TimedCall) { auto it = m_timerQueue.begin(); while (it != m_timerQueue.end() && (*it)->TargetTime < item->TargetTime) { @@ -91,7 +188,6 @@ namespace PAL_NS_BEGIN { else { m_queue.push_back(item); } - count++; m_event.post(); } @@ -128,14 +224,14 @@ namespace PAL_NS_BEGIN { return false; } - if (m_itemInProgress == item) + if (m_itemInProgress.load(std::memory_order_acquire) == item) { /* Can't recursively wait on completion of our own thread */ if (m_hThread.get_id() != std::this_thread::get_id()) { if (waitTime > 0 && m_execution_mutex.try_lock_for(std::chrono::milliseconds(waitTime))) { - m_itemInProgress = nullptr; + m_itemInProgress.store(nullptr, std::memory_order_release); m_execution_mutex.unlock(); } } @@ -150,7 +246,7 @@ namespace PAL_NS_BEGIN { * true - if item in progress is different than item (other task) * false - if item in progress is still the same (didn't wait long enough) */ - return (m_itemInProgress != item); + return (m_itemInProgress.load(std::memory_order_acquire) != item); } { @@ -181,7 +277,11 @@ namespace PAL_NS_BEGIN { uint64_t wakeupCount = 0; WorkerThread* self = reinterpret_cast(lpThreadParameter); - LOG_INFO("Running thread %u", std::this_thread::get_id()); + { + LOCKGUARD(self->m_lock); + self->m_workerId = std::this_thread::get_id(); + } + LOG_INFO("Running thread %zu", std::hash{}(std::this_thread::get_id())); for (;;) { std::unique_ptr item = nullptr; @@ -218,7 +318,7 @@ namespace PAL_NS_BEGIN { } if (item) { - self->m_itemInProgress = item.get(); + self->m_itemInProgress.store(item.get(), std::memory_order_release); } } @@ -230,7 +330,17 @@ namespace PAL_NS_BEGIN { if (item->Type == MAT::Task::Shutdown) { item.reset(); - self->m_itemInProgress = nullptr; + self->m_itemInProgress.store(nullptr, std::memory_order_release); + // Drop any tasks still queued behind the shutdown sentinel + // (e.g. future-dated timers) before exiting. The owning thread + // deletes these in Join() only after a successful join(); on the + // self-Join path it detaches and skips that cleanup, so draining + // here prevents leaking those tasks. This matches the join()-path + // behavior of dropping un-run work at shutdown. + { + LOCKGUARD(self->m_lock); + self->drainPendingTasksLocked(); + } break; } @@ -238,7 +348,7 @@ namespace PAL_NS_BEGIN { std::lock_guard lock(self->m_execution_mutex); // Item wasn't cancelled before it could be executed - if (self->m_itemInProgress != nullptr) { + if (self->m_itemInProgress.load(std::memory_order_acquire) != nullptr) { LOG_TRACE("%10llu Execute item=%p type=%s\n", wakeupCount, item.get(), item.get()->TypeName.c_str() ); // A task can run arbitrary work (storage I/O, HTTP encode, and // user DebugEventListener callbacks). An exception escaping here @@ -253,7 +363,7 @@ namespace PAL_NS_BEGIN { catch (...) { LOG_ERROR("Unhandled non-standard exception in worker task"); } - self->m_itemInProgress = nullptr; + self->m_itemInProgress.store(nullptr, std::memory_order_release); } if (item) { @@ -262,17 +372,31 @@ namespace PAL_NS_BEGIN { } } } + + // The loop has broken on a Shutdown item. If the last reference was + // released by a task on this worker thread, onLastReferenceReleased() + // detached and deferred deletion to us; perform it now, after all member + // access is done, so the object outlives threadFunc rather than being + // freed underneath it. + if (self->m_disposeFromThread.load(std::memory_order_acquire)) { + delete self; + } } }; namespace WorkerThreadFactory { std::shared_ptr Create() { - return std::make_shared(); + // Custom deleter so that a last-reference release happening on the worker + // thread itself defers destruction to the thread (see + // onLastReferenceReleased) instead of freeing the object underneath a + // still-running threadFunc. + return std::shared_ptr( + new WorkerThread(), + [](WorkerThread* self) { if (self->onLastReferenceReleased()) delete self; }); } } } PAL_NS_END #endif - diff --git a/lib/tpm/TransmissionPolicyManager.cpp b/lib/tpm/TransmissionPolicyManager.cpp index 83b82cf2a..720ad344a 100644 --- a/lib/tpm/TransmissionPolicyManager.cpp +++ b/lib/tpm/TransmissionPolicyManager.cpp @@ -111,26 +111,36 @@ namespace MAT_NS_BEGIN { LOG_TRACE("Collector URL is not set, no upload."); return; } - LOCKGUARD(m_scheduledUploadMutex); - if (delay.count() < 0 || m_timerdelay.count() < 0) - { - LOG_TRACE("Negative delay(%d) or m_timerdelay(%d), no upload", delay.count(), m_timerdelay.count()); - return; - } - if (m_scheduledUploadAborted) - { - LOG_TRACE("Scheduled upload aborted, no upload."); - return; - } - if (uploadCount() >= static_cast(m_config[CFG_INT_MAX_PENDING_REQ]) ) + auto shouldSkipScheduling = [&delay, this]() -> bool { - LOG_TRACE("Maximum number of HTTP requests reached"); - return; - } + if (delay.count() < 0 || m_timerdelay.count() < 0) + { + LOG_TRACE("Negative delay(%lld) or m_timerdelay(%lld), no upload", + static_cast(delay.count()), static_cast(m_timerdelay.count())); + return true; + } + if (m_scheduledUploadAborted) + { + LOG_TRACE("Scheduled upload aborted, no upload."); + return true; + } + if (uploadCount() >= static_cast(m_config[CFG_INT_MAX_PENDING_REQ])) + { + LOG_TRACE("Maximum number of HTTP requests reached"); + return true; + } + if (m_isPaused) + { + LOG_TRACE("Paused, not uploading anything until resumed"); + return true; + } + + return false; + }; - if (m_isPaused) + LOCKGUARD(m_scheduledUploadMutex); + if (shouldSkipScheduling()) { - LOG_TRACE("Paused, not uploading anything until resumed"); return; } @@ -151,10 +161,9 @@ namespace MAT_NS_BEGIN { if (delta <= static_cast(delay.count())) { // Don't need to cancel and reschedule if it's about to happen now anyways. - // m_isUploadScheduled check does not have to be strictly atomic because // the completion of upload will schedule more uploads as-needed, we only // want to avoid the unnecessary wasteful rescheduling. - LOG_TRACE("WAIT upload %d ms for lat=%d", delta, m_runningLatency); + LOG_TRACE("WAIT upload %llu ms for lat=%d", static_cast(delta), m_runningLatency); return; } } @@ -162,18 +171,35 @@ namespace MAT_NS_BEGIN { // Cancel upload if already scheduled. if (force || delay.count() == 0) { - if (!cancelUploadTask()) + if (!cancelUploadTaskNoWaitLocked()) { LOG_TRACE("Upload either hasn't been scheduled or already done."); + // Cancel can return false when the previous upload task is + // currently executing on the worker. If uploadAsync hasn't + // yet entered its own LOCKGUARD (m_isUploadScheduled is + // still set under the mutex we hold), propagate the + // requested latency so the running task picks it up when + // it acquires m_scheduledUploadMutex. Otherwise the + // running task has already cleared the flag and the + // schedule below will queue a fresh task. + if (m_isUploadScheduled) + { + m_runningLatency = latency; + } + } + if (shouldSkipScheduling()) + { + return; } } // Schedule new upload - if (!m_isUploadScheduled.exchange(true)) + if (!m_isUploadScheduled) { + m_isUploadScheduled = true; m_scheduledUploadTime = PAL::getMonotonicTimeMs() + delay.count(); m_runningLatency = latency; - LOG_TRACE("SCHED upload %d ms for lat=%d", delay.count(), m_runningLatency); + LOG_TRACE("SCHED upload %lld ms for lat=%d", static_cast(delay.count()), m_runningLatency); m_scheduledUpload = PAL::scheduleTask(&m_taskDispatcher, static_cast(delay.count()), this, &TransmissionPolicyManager::uploadAsync, latency); } } @@ -184,16 +210,15 @@ namespace MAT_NS_BEGIN { if (guard.isPaused()) { return; } - m_runningLatency = latency; - m_scheduledUploadTime = std::numeric_limits::max(); - + EventLatency requestedLatency = latency; { LOCKGUARD(m_scheduledUploadMutex); + requestedLatency = m_runningLatency; + m_scheduledUploadTime = std::numeric_limits::max(); m_isUploadScheduled = false; // Allow to schedule another uploadAsync if ((m_isPaused) || (m_scheduledUploadAborted)) { - LOG_TRACE("Paused or upload aborted: cancel pending upload task."); - cancelUploadTask(); // If there is a pending upload task, kill it + LOG_TRACE("Paused or upload aborted: skip upload."); return; } } @@ -210,14 +235,14 @@ namespace MAT_NS_BEGIN { unsigned delayMs = 1000; LOG_INFO("Bandwidth controller proposed bandwidth %u bytes/sec but minimum accepted is %u, will retry %u ms later", proposedBandwidthBps, minimumBandwidthBps, delayMs); - scheduleUpload(delayMs, latency); // reschedule uploadAsync to run again 1000 ms later + scheduleUpload(std::chrono::milliseconds{delayMs}, requestedLatency); // reschedule uploadAsync to run again 1000 ms later return; } } #endif auto ctx = m_system.createEventsUploadContext(); - ctx->requestedMinLatency = m_runningLatency; + ctx->requestedMinLatency = requestedLatency; addUpload(ctx); initiateUpload(ctx); } @@ -238,7 +263,7 @@ namespace MAT_NS_BEGIN { // Rescheduling upload if (nextUpload.count() >= 0) { - LOG_TRACE("Scheduling upload in %d ms", nextUpload.count()); + LOG_TRACE("Scheduling upload in %lld ms", static_cast(nextUpload.count())); EventLatency proposed = calculateNewPriority(); scheduleUpload(nextUpload, proposed); // reschedule uploadAsync again } @@ -284,9 +309,9 @@ namespace MAT_NS_BEGIN { LOCKGUARD(m_scheduledUploadMutex); // Prevent execution of all upload tasks m_scheduledUploadAborted = true; - // Make sure we wait for completion of the upload scheduling task that may be running - cancelUploadTask(); } + // Make sure we wait for completion of the upload scheduling task that may be running + cancelUploadTask(); // Make sure we wait for all active upload callbacks to finish while (uploadCount() > 0) @@ -342,7 +367,12 @@ namespace MAT_NS_BEGIN { } // Schedule async upload if not scheduled yet - if (!m_isUploadScheduled || TransmitProfiles::isTimerUpdateRequired()) + bool isUploadScheduled = false; + { + LOCKGUARD(m_scheduledUploadMutex); + isUploadScheduled = m_isUploadScheduled; + } + if (!isUploadScheduled || TransmitProfiles::isTimerUpdateRequired()) { if (updateTimersIfNecessary()) { @@ -374,7 +404,13 @@ namespace MAT_NS_BEGIN { return EventLatency_RealTime; } - if (m_runningLatency == EventLatency_RealTime) + EventLatency runningLatency = EventLatency_RealTime; + { + LOCKGUARD(m_scheduledUploadMutex); + runningLatency = m_runningLatency; + } + + if (runningLatency == EventLatency_RealTime) { return EventLatency_Normal; } @@ -453,16 +489,39 @@ namespace MAT_NS_BEGIN { return (m_scheduledUploadAborted) ? DefaultTaskCancelTime : std::chrono::milliseconds {}; } + bool TransmissionPolicyManager::cancelUploadTaskNoWaitLocked() + { + bool result = m_scheduledUpload.Cancel(std::chrono::milliseconds {}.count()); + + if (result) + { + m_isUploadScheduled = false; + m_scheduledUploadTime = std::numeric_limits::max(); + } + return result; + } + bool TransmissionPolicyManager::cancelUploadTask() { - bool result = m_scheduledUpload.Cancel(getCancelWaitTime().count()); + auto waitTime = std::chrono::milliseconds{}; + { + LOCKGUARD(m_scheduledUploadMutex); + waitTime = getCancelWaitTime(); + if (waitTime.count() == 0) + { + return cancelUploadTaskNoWaitLocked(); + } + } + bool result = m_scheduledUpload.Cancel(waitTime.count()); // TODO: There is a potential for upload tasks to not be canceled, especially if they aren't waited for. // We either need a stronger guarantee here (could impact SDK performance), or a mechanism to // ensure those tasks are canceled when the log manager is destroyed. Issue 388 if (result) { - m_isUploadScheduled.exchange(false); + LOCKGUARD(m_scheduledUploadMutex); + m_isUploadScheduled = false; + m_scheduledUploadTime = std::numeric_limits::max(); } return result; } @@ -473,9 +532,10 @@ namespace MAT_NS_BEGIN { return m_activeUploads.size(); } - bool TransmissionPolicyManager::isUploadInProgress() const noexcept + bool TransmissionPolicyManager::isUploadInProgress() const { // unfinished uploads that haven't processed callbacks or pending upload task + LOCKGUARD(m_scheduledUploadMutex); return (uploadCount() > 0) || m_isUploadScheduled; } diff --git a/lib/tpm/TransmissionPolicyManager.hpp b/lib/tpm/TransmissionPolicyManager.hpp index e1a91ad10..dd69a6e52 100644 --- a/lib/tpm/TransmissionPolicyManager.hpp +++ b/lib/tpm/TransmissionPolicyManager.hpp @@ -90,9 +90,9 @@ constexpr const char* const DefaultBackoffConfig = "E,3000,300000,2,1"; DeviceStateHandler m_deviceStateHandler; std::atomic m_isPaused { true }; - std::atomic m_isUploadScheduled { false }; + bool m_isUploadScheduled { false }; uint64_t m_scheduledUploadTime { std::numeric_limits::max() }; - std::mutex m_scheduledUploadMutex; + mutable std::mutex m_scheduledUploadMutex; PAL::DeferredCallbackHandle m_scheduledUpload; bool m_scheduledUploadAborted { false }; @@ -119,6 +119,12 @@ constexpr const char* const DefaultBackoffConfig = "E,3000,300000,2,1"; std::chrono::milliseconds getCancelWaitTime() const noexcept; + /// + /// Cancels a pending upload task without waiting for a running task to finish. + /// The caller must already hold m_scheduledUploadMutex. + /// + bool cancelUploadTaskNoWaitLocked(); + /// /// Cancels pending upload task. /// @@ -152,7 +158,7 @@ constexpr const char* const DefaultBackoffConfig = "E,3000,300000,2,1"; RouteSink eventsUploadFailed{ this, &TransmissionPolicyManager::handleEventsUploadFailed }; RouteSink eventsUploadAborted{ this, &TransmissionPolicyManager::handleEventsUploadAborted }; - virtual bool isUploadInProgress() const noexcept; + virtual bool isUploadInProgress() const; virtual bool isPaused() const noexcept; }; @@ -160,4 +166,3 @@ constexpr const char* const DefaultBackoffConfig = "E,3000,300000,2,1"; } MAT_NS_END #endif // TRANSMISSIONPOLICYMANAGER_HPP - diff --git a/lib/tpm/TransmitProfiles.cpp b/lib/tpm/TransmitProfiles.cpp index 5daec5f8b..b26766f6f 100644 --- a/lib/tpm/TransmitProfiles.cpp +++ b/lib/tpm/TransmitProfiles.cpp @@ -58,6 +58,7 @@ static void initTransmitProfileFields() transmitProfilePowerState["unknown"] = (PowerSource_Unknown); transmitProfilePowerState["battery"] = (PowerSource_Battery); transmitProfilePowerState["charging"] = (PowerSource_Charging); + transmitProfilePowerState["low_battery"] = (PowerSource_LowBattery); }; #endif @@ -103,11 +104,14 @@ namespace MAT_NS_BEGIN { LOG_TRACE("name=%s", profile.name.c_str()); size_t i = 0; for (auto &rule : profile.rules) { - LOG_TRACE("[%d] netCost=%2d, powState=%2d, timers=[%3d,%3d,%3d]", + // Custom profiles may supply fewer than three timers, so read + // out-of-range slots as 0 instead of indexing past the vector. + auto timerOrZero = [&rule](size_t idx) { return idx < rule.timers.size() ? rule.timers[idx] : 0; }; + LOG_TRACE("[%zu] netCost=%2d, powState=%2d, timers=[%3d,%3d,%3d]", i, rule.netCost, rule.powerState, - rule.timers[0], - rule.timers[1], - rule.timers[2]); + timerOrZero(0), + timerOrZero(1), + timerOrZero(2)); i++; } } @@ -512,14 +516,17 @@ namespace MAT_NS_BEGIN { isTimerUpdated = true; #ifdef HAVE_MAT_LOGGING auto it = profiles.find(currProfileName); - if (it != profiles.end()) { + if (it != profiles.end() && currRule < it->second.rules.size()) { /* Debug routine to print the list of currently selected timers */ TransmitProfileRule &rule = (it->second).rules[currRule]; + // The rule may carry fewer than three timers, so read out-of-range + // slots as 0 instead of indexing past the vector. + auto timerOrZero = [&rule](size_t idx) { return idx < rule.timers.size() ? rule.timers[idx] : 0; }; // Print just 3 timers for now because we support only 3 LOG_INFO("timers=[%3d,%3d,%3d]", - rule.timers[0], - rule.timers[1], - rule.timers[2]); + timerOrZero(0), + timerOrZero(1), + timerOrZero(2)); } #endif } diff --git a/lib/utils/Utils.cpp b/lib/utils/Utils.cpp index 22a48d87f..b8ac0f6cd 100644 --- a/lib/utils/Utils.cpp +++ b/lib/utils/Utils.cpp @@ -103,15 +103,30 @@ namespace MAT_NS_BEGIN { if (IsRunningInApp()) { auto hr = RoInitialize(RO_INIT_MULTITHREADED); - /* Ignoring result from call to `RoInitialize` as either initialzation is successful, or else already - * initialized and it should be ok to proceed in both the scenarios */ - UNREFERENCED_PARAMETER(hr); - - ::Windows::Storage::StorageFolder ^ temp = ::Windows::Storage::ApplicationData::Current->TemporaryFolder; - // TODO: [MG] - // - verify that the path ends with a slash - // -- add exception handler in case if AppData temp folder is not accessible - return from_platform_string(temp->Path->ToString()); + // RoInitialize returns S_OK when it initializes the apartment and + // S_FALSE when it was already initialized on this thread; both add a + // reference that must be balanced with RoUninitialize. The RAII guard + // balances a successful init on every exit path, including if a WinRT + // call below throws. RPC_E_CHANGED_MODE and other failures did not + // initialize and are left unbalanced. + struct ApartmentGuard + { + HRESULT hr; + ~ApartmentGuard() { if (SUCCEEDED(hr)) { RoUninitialize(); } } + } apartmentGuard{hr}; + + std::string tempPath; + { + // Release the WinRT StorageFolder before the guard runs (at the + // end of the enclosing scope) so the object is not destroyed in an + // uninitialized apartment. + ::Windows::Storage::StorageFolder ^ temp = ::Windows::Storage::ApplicationData::Current->TemporaryFolder; + // TODO: [MG] + // - verify that the path ends with a slash + // -- add exception handler in case if AppData temp folder is not accessible + tempPath = from_platform_string(temp->Path->ToString()); + } + return tempPath; } else { diff --git a/tests/functests/BasicFuncTests.cpp b/tests/functests/BasicFuncTests.cpp index bc879d3e6..23ca3b76c 100644 --- a/tests/functests/BasicFuncTests.cpp +++ b/tests/functests/BasicFuncTests.cpp @@ -186,7 +186,7 @@ class BasicFuncTests : public ::testing::Test, std::remove((fileName + "-journal").c_str()); } - virtual void Initialize() + virtual void Initialize(int64_t maxTeardownUploadTimeInSec = 2) { receivedRequests.clear(); auto configuration = LogManager::GetLogConfiguration(); @@ -202,14 +202,14 @@ class BasicFuncTests : public ::testing::Test, configuration[CFG_INT_RAM_QUEUE_SIZE] = 4096 * 20; configuration[CFG_STR_CACHE_FILE_PATH] = TEST_STORAGE_FILENAME; configuration[CFG_INT_CACHE_FILE_SIZE] = 4096 * 1024; // 4MB default - configuration[CFG_INT_MAX_TEARDOWN_TIME] = 2; // 2 seconds wait on shutdown + configuration[CFG_INT_MAX_TEARDOWN_TIME] = maxTeardownUploadTimeInSec; configuration[CFG_INT_STORAGE_FULL_PCT] = 75; // default configuration[CFG_INT_STORAGE_FULL_CHECK_TIME] = 5000; // default 5s configuration[CFG_STR_COLLECTOR_URL] = serverAddress.c_str(); configuration[CFG_MAP_HTTP][CFG_BOOL_HTTP_COMPRESSION] = false; // disable compression for now configuration[CFG_MAP_TPM][CFG_STR_TPM_BACKOFF] = "E,500,5000,2,1"; // faster retry for localhost tests configuration[CFG_MAP_METASTATS_CONFIG][CFG_INT_METASTATS_INTERVAL] = 30 * 60; // 30 mins - configuration[CFG_MAP_METASTATS_CONFIG]["enabled"] = true; // opt in to stats (disabled by default since #1420) + configuration[CFG_MAP_METASTATS_CONFIG]["enabled"] = true; // opt in to stats (disabled by default) configuration["name"] = __FILE__; configuration["version"] = "1.0.0"; @@ -594,6 +594,45 @@ TEST_F(BasicFuncTests, sendOneEvent_immediatelyStop) EXPECT_GE(receivedRequests.size(), (size_t)1); // at least 1 HTTP request with customer payload and stats } +TEST_F(BasicFuncTests, teardownDuringInFlightUpload_ShutsDownCleanly) +{ + // Smoke test for teardown while an upload is in flight. + // Uploads target the /slow/ endpoint with large payloads and MAX_TEARDOWN_TIME + // is 0, so FlushAndTeardown() returns while an upload is still outstanding. + // Teardown must complete cleanly without touching freed SDK state; run under a + // sanitizer (ASan/TSan) this guards the teardown-vs-upload path. + CleanStorage(); + static int64_t const ONE_EVENT_SIZE = 256 * 1024; + + // Point Initialize() at the (slow) endpoint so uploads stay in flight. + std::string savedAddress = serverAddress; + size_t pos = serverAddress.rfind("/simple/"); + // Assert the rewrite actually happens: if the base URL format ever changes and + // no longer contains "/simple/", uploads would hit the normal endpoint and the + // in-flight teardown scenario would not be exercised, yet the test would still + // pass. Fail loudly instead so the regression coverage can't silently lapse. + ASSERT_NE(pos, std::string::npos) + << "serverAddress '" << serverAddress << "' does not contain '/simple/'; " + << "the /slow/ rewrite would be a no-op and this test would not exercise " + << "teardown during an in-flight upload."; + serverAddress.replace(pos, std::string("/simple/").size(), "/slow/"); + Initialize(0); + serverAddress = savedAddress; + + for (int i = 0; i < 20; ++i) + { + EventProperties event("teardown_event"); + event.SetPriority(EventPriority_Normal); + event.SetProperty("big_data", std::string(static_cast(ONE_EVENT_SIZE), 'x')); + logger->LogEvent(event); + } + LogManager::UploadNow(); + PAL::sleep(300); // let the upload reach the slow server so it is in flight + // Teardown with timeout 0 returns while the upload is still outstanding. + LogManager::FlushAndTeardown(); + SUCCEED(); +} + TEST_F(BasicFuncTests, sendNoPriorityEvents) { CleanStorage(); @@ -1200,7 +1239,7 @@ TEST_F(BasicFuncTests, killSwitchWorks) 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_METASTATS_CONFIG]["enabled"] = true; // opt in to stats (disabled by default) configuration["name"] = __FILE__; configuration["version"] = "1.0.0"; @@ -1280,9 +1319,9 @@ TEST_F(BasicFuncTests, killIsTemporary) configuration[CFG_STR_CACHE_FILE_PATH] = TEST_STORAGE_FILENAME; configuration[CFG_INT_MAX_TEARDOWN_TIME] = 2; configuration[CFG_STR_COLLECTOR_URL] = serverAddress.c_str(); - 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[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) configuration["name"] = __FILE__; configuration["version"] = "1.0.0"; configuration["config"] = { { "host", __FILE__ } }; diff --git a/tests/unittests/HttpResponseDecoderTests.cpp b/tests/unittests/HttpResponseDecoderTests.cpp index 314cdb513..7d11ae4b8 100644 --- a/tests/unittests/HttpResponseDecoderTests.cpp +++ b/tests/unittests/HttpResponseDecoderTests.cpp @@ -88,20 +88,29 @@ TEST_F(HttpResponseDecoderTests, UnderstandsTemporaryServerFailures) TEST_F(HttpResponseDecoderTests, UnderstandsTemporaryNetworkFailures) { auto ctx = createContextWith(HttpResult_LocalFailure, -1, ""); - EXPECT_CALL(*this, resultTemporaryNetworkFailure(ctx)) - .WillOnce(Return()); + EXPECT_CALL(*this, resultTemporaryNetworkFailure(ctx)).WillOnce(Invoke([](EventsUploadContextPtr const& routedCtx) { + ASSERT_THAT(routedCtx->httpResponse, NotNull()); + EXPECT_THAT(routedCtx->httpResponse->GetResult(), HttpResult_LocalFailure); + EXPECT_THAT(routedCtx->httpResponse->GetStatusCode(), static_cast(-1)); + })); decoder.decode(ctx); ctx = createContextWith(HttpResult_NetworkFailure, -1, ""); - EXPECT_CALL(*this, resultTemporaryNetworkFailure(ctx)) - .WillOnce(Return()); + EXPECT_CALL(*this, resultTemporaryNetworkFailure(ctx)).WillOnce(Invoke([](EventsUploadContextPtr const& routedCtx) { + ASSERT_THAT(routedCtx->httpResponse, NotNull()); + EXPECT_THAT(routedCtx->httpResponse->GetResult(), HttpResult_NetworkFailure); + EXPECT_THAT(routedCtx->httpResponse->GetStatusCode(), static_cast(-1)); + })); decoder.decode(ctx); } TEST_F(HttpResponseDecoderTests, SkipsAbortedRequests) { auto ctx = createContextWith(HttpResult_Aborted, -1, ""); - EXPECT_CALL(*this, resultRequestAborted(ctx)) - .WillOnce(Return()); + EXPECT_CALL(*this, resultRequestAborted(ctx)).WillOnce(Invoke([](EventsUploadContextPtr const& routedCtx) { + ASSERT_THAT(routedCtx->httpResponse, NotNull()); + EXPECT_THAT(routedCtx->httpResponse->GetResult(), HttpResult_Aborted); + EXPECT_THAT(routedCtx->httpResponse->GetStatusCode(), static_cast(-1)); + })); decoder.decode(ctx); } diff --git a/tests/unittests/MemoryStorageTests.cpp b/tests/unittests/MemoryStorageTests.cpp index a736d125f..268cf137d 100644 --- a/tests/unittests/MemoryStorageTests.cpp +++ b/tests/unittests/MemoryStorageTests.cpp @@ -213,6 +213,24 @@ TEST_F(MemoryStorageTests, DeleteAllRecords) EXPECT_THAT(storage.GetReservedCount(), 0); } +TEST_F(MemoryStorageTests, DeleteRecordsWithEmptyFilterDoesNotDeleteAll) +{ + MemoryStorage storage(testLogManager, *testConfig); + + // Add some events to storage + auto total_db_size = addEvents(storage); + EXPECT_THAT(storage.GetSize(), total_db_size); + auto count_before = storage.GetRecordCount(); + EXPECT_GT(count_before, static_cast(0)); + + // An empty where-filter matches every record; it must NOT wipe the queue. + // Intentional full clears go through DeleteAllRecords(). + storage.DeleteRecords(std::map{}); + + EXPECT_THAT(storage.GetRecordCount(), count_before); + EXPECT_THAT(storage.GetSize(), total_db_size); +} + TEST_F(MemoryStorageTests, ReleaseRecords) { diff --git a/tests/unittests/OfflineStorageTests.cpp b/tests/unittests/OfflineStorageTests.cpp index bbb8da8e0..581b4be6a 100644 --- a/tests/unittests/OfflineStorageTests.cpp +++ b/tests/unittests/OfflineStorageTests.cpp @@ -2,7 +2,15 @@ #include "common/Common.hpp" #include "common/MockIOfflineStorage.hpp" +#include "common/MockIOfflineStorageObserver.hpp" +#include "common/MockIRuntimeConfig.hpp" +#include "offline/OfflineStorageHandler.hpp" #include "offline/StorageObserver.hpp" +#include "NullObjects.hpp" + +#include +#include +#include using namespace testing; using namespace MAT; @@ -162,3 +170,305 @@ TEST_F(OfflineStorageTests, ReleaseRecordsIsForwarded) .WillOnce(Return()); EXPECT_THAT(offlineStorage.releaseRecordsIncRetryCount(ctx), true); } + +namespace +{ + // Remove a SQLite db file along with its WAL-mode companion files + // (-wal/-shm/-journal), which would otherwise accumulate in the temp dir. + void RemoveDbFiles(const std::string& path) + { + std::remove(path.c_str()); + std::remove((path + "-wal").c_str()); + std::remove((path + "-shm").c_str()); + std::remove((path + "-journal").c_str()); + } + + // No-op dispatcher that owns queued tasks and frees them, so flushes only + // run when invoked directly and scheduled tasks (if any) are not leaked. + class NoopTaskDispatcher : public ITaskDispatcher + { + public: + void Join() override { clear(); } + void Queue(Task* task) override { m_tasks.push_back(task); } + bool Cancel(Task* task, uint64_t waitTime = 0) override + { + UNREFERENCED_PARAMETER(waitTime); + auto it = std::find(m_tasks.begin(), m_tasks.end(), task); + if (it != m_tasks.end()) + { + delete *it; + m_tasks.erase(it); + return true; + } + return false; + } + ~NoopTaskDispatcher() override { clear(); } + + private: + void clear() + { + for (auto* t : m_tasks) + delete t; + m_tasks.clear(); + } + std::vector m_tasks; + }; +} + +namespace MAT_NS_BEGIN { + + class OfflineStorageHandlerTestPeer + { + public: + static void SetObserver(OfflineStorageHandler& handler, IOfflineStorageObserver& observer) + { + handler.m_observer = &observer; + } + + static void SetMemoryStorage(OfflineStorageHandler& handler, IOfflineStorage* storage) + { + handler.m_offlineStorageMemory.reset(storage); + } + + static void SetDiskStorage(OfflineStorageHandler& handler, std::shared_ptr storage) + { + handler.m_offlineStorageDisk = storage; + } + + static size_t ReturnRecordsToMemory(OfflineStorageHandler& handler, std::vector const& records) + { + return handler.ReturnRecordsToMemory(records); + } + }; + +} MAT_NS_END + +TEST(OfflineStorageHandlerFlushTests, FailedMemoryRequeueIsReportedAndDropped) +{ + NullLogManager logManager; + NiceMock config; + NoopTaskDispatcher dispatcher; + StrictMock observer; + + OfflineStorageHandler handler(logManager, config, dispatcher); + OfflineStorageHandlerTestPeer::SetObserver(handler, observer); + + auto* memory = new StrictMock(); + OfflineStorageHandlerTestPeer::SetMemoryStorage(handler, memory); + + std::vector records; + records.push_back(StorageRecord("retry-ok", "tenant-one-token", + EventLatency_Normal, EventPersistence_Normal, /*timestamp*/ 1, + std::vector{ 'x' })); + records.push_back(StorageRecord("retry-drop", "tenant-two-token", + EventLatency_Normal, EventPersistence_Normal, /*timestamp*/ 1, + std::vector{ 'y' })); + + EXPECT_CALL(*memory, StoreRecord(_)) + .WillOnce(Return(true)) + .WillOnce(Return(false)); + EXPECT_CALL(observer, OnStorageRecordsDropped(_)) + .WillOnce(Invoke([](std::map const& dropped) { + auto found = dropped.find("tenant-two-token"); + ASSERT_NE(found, dropped.end()); + EXPECT_EQ(found->second, static_cast(1)); + })); + + EXPECT_EQ(OfflineStorageHandlerTestPeer::ReturnRecordsToMemory(handler, records), + static_cast(1)); +} + +TEST(OfflineStorageHandlerFlushTests, BatchingOptOutUsesPerRecordDiskStores) +{ + NullLogManager logManager; + NiceMock config; + NoopTaskDispatcher dispatcher; + StrictMock observer; + + config[CFG_BOOL_ENABLE_BATCHED_STORAGE_FLUSH] = false; + + OfflineStorageHandler handler(logManager, config, dispatcher); + OfflineStorageHandlerTestPeer::SetObserver(handler, observer); + + auto* memory = new StrictMock(); + std::shared_ptr> disk(new StrictMock()); + OfflineStorageHandlerTestPeer::SetMemoryStorage(handler, memory); + OfflineStorageHandlerTestPeer::SetDiskStorage(handler, disk); + + std::vector records; + records.push_back(StorageRecord("per-record-1", "tenant-one-token", + EventLatency_Normal, EventPersistence_Normal, /*timestamp*/ 1, + std::vector{ 'x' })); + records.push_back(StorageRecord("per-record-2", "tenant-two-token", + EventLatency_Normal, EventPersistence_Normal, /*timestamp*/ 1, + std::vector{ 'y' })); + + EXPECT_CALL(*memory, GetSize()) + .WillOnce(Return(static_cast(records.size()))) + .WillOnce(Return(static_cast(0))); + EXPECT_CALL(*memory, GetRecords(false, EventLatency_Unspecified, 0)) + .WillOnce(Return(records)); + EXPECT_CALL(*disk, StoreRecords(_)).Times(0); + EXPECT_CALL(*disk, StoreRecord(_)) + .Times(static_cast(records.size())) + .WillRepeatedly(Return(true)); + EXPECT_CALL(observer, OnStorageRecordsSaved(records.size())); + + handler.Flush(); +} + +// Regression test: when valid records drained from the in-memory queue fail to +// be persisted by the disk backend during Flush() (a transient failure -- here +// an unopenable database), they must be returned to the queue rather than lost. +TEST(OfflineStorageHandlerFlushTests, FailedDiskStoreDuringFlushReturnsRecordsToMemory) +{ + NullLogManager logManager; + NiceMock config; + NoopTaskDispatcher dispatcher; + NiceMock observer; + + ON_CALL(config, GetOfflineStorageMaximumSizeBytes()).WillByDefault(Return(32 * 4096)); + ON_CALL(config, GetMaximumRetryCount()).WillByDefault(Return(5)); + + // A path inside a non-existent directory cannot be opened by SQLite (it does + // not create parent directories), so every disk StoreRecords() returns 0 -- + // a transient failure with otherwise-valid records. + std::ostringstream dbPath; + dbPath << GetTempDirectory() << "no_such_dir_" << PAL::getUtcSystemTimeMs() + << "/FlushReserveTest.db"; + config[CFG_STR_CACHE_FILE_PATH] = dbPath.str(); + config[CFG_INT_RAM_QUEUE_SIZE] = 1024 * 1024; // enable the in-memory queue + + OfflineStorageHandler handler(logManager, config, dispatcher); + handler.Initialize(observer); + + const size_t kCount = 5; + for (size_t i = 0; i < kCount; i++) + { + StorageRecord r("flush-id-" + std::to_string(i), "tenant-token", + EventLatency_Normal, EventPersistence_Normal, /*timestamp*/ 1, + std::vector{ 'x' }); + handler.StoreRecord(r); + } + EXPECT_EQ(handler.GetRecordCount(), kCount); + + handler.Flush(); + + // The disk could not persist the batch; with the fix the valid records are + // returned to the in-memory queue rather than silently dropped. + EXPECT_EQ(handler.GetRecordCount(), kCount); + + handler.Shutdown(); +} + +TEST(OfflineStorageHandlerFlushTests, EventLatencyOffIsDroppedWithoutReportingStoreFailure) +{ + NullLogManager logManager; + NiceMock config; + NoopTaskDispatcher dispatcher; + NiceMock observer; + + ON_CALL(config, GetOfflineStorageMaximumSizeBytes()).WillByDefault(Return(32 * 4096)); + + std::ostringstream dbPath; + dbPath << GetTempDirectory() << "LatencyOff-" << PAL::getUtcSystemTimeMs() << ".db"; + RemoveDbFiles(dbPath.str()); + config[CFG_STR_CACHE_FILE_PATH] = dbPath.str(); + config[CFG_INT_RAM_QUEUE_SIZE] = 1024 * 1024; // enable the in-memory queue + + OfflineStorageHandler handler(logManager, config, dispatcher); + handler.Initialize(observer); + + StorageRecord record("latency-off", "tenant-token", + EventLatency_Off, EventPersistence_Normal, /*timestamp*/ 1, + std::vector{ 'x' }); + + EXPECT_TRUE(handler.StoreRecord(record)); + EXPECT_EQ(handler.GetRecordCount(), static_cast(0)); + + handler.Shutdown(); + RemoveDbFiles(dbPath.str()); +} + +// Regression test: a permanently-invalid record (rejected by the disk backend's +// validation) must be dropped on Flush(), not returned to the queue -- otherwise +// one poison record would be re-drained and re-rejected on every flush, wedging +// the queue and blocking every valid record behind it. +TEST(OfflineStorageHandlerFlushTests, FlushDropsInvalidRecordsInsteadOfWedging) +{ + NullLogManager logManager; + NiceMock config; + NoopTaskDispatcher dispatcher; + NiceMock observer; + + ON_CALL(config, GetOfflineStorageMaximumSizeBytes()).WillByDefault(Return(32 * 4096)); + ON_CALL(config, GetMaximumRetryCount()).WillByDefault(Return(5)); + + std::ostringstream dbPath; + dbPath << GetTempDirectory() << "FlushDropInvalid-" << PAL::getUtcSystemTimeMs() << ".db"; + RemoveDbFiles(dbPath.str()); + config[CFG_STR_CACHE_FILE_PATH] = dbPath.str(); + config[CFG_INT_RAM_QUEUE_SIZE] = 1024 * 1024; // enable the in-memory queue + + OfflineStorageHandler handler(logManager, config, dispatcher); + handler.Initialize(observer); + + // A timestamp <= 0 is accepted by the in-memory queue but permanently rejected + // by the SQLite disk store's validation, so it can never be persisted. + const size_t kCount = 5; + for (size_t i = 0; i < kCount; i++) + { + StorageRecord r("bad-id-" + std::to_string(i), "tenant-token", + EventLatency_Normal, EventPersistence_Normal, /*timestamp*/ 0, + std::vector{ 'x' }); + handler.StoreRecord(r); + } + EXPECT_EQ(handler.GetRecordCount(), kCount); + + handler.Flush(); + + // The invalid records are dropped, not returned to the queue, so the queue + // drains and is not wedged. + EXPECT_EQ(handler.GetRecordCount(), static_cast(0)); + + handler.Shutdown(); + RemoveDbFiles(dbPath.str()); +} + +TEST(OfflineStorageHandlerFlushTests, FlushOptOutDropsInvalidRecordsInsteadOfWedging) +{ + NullLogManager logManager; + NiceMock config; + NoopTaskDispatcher dispatcher; + NiceMock observer; + + ON_CALL(config, GetOfflineStorageMaximumSizeBytes()).WillByDefault(Return(32 * 4096)); + ON_CALL(config, GetMaximumRetryCount()).WillByDefault(Return(5)); + + std::ostringstream dbPath; + dbPath << GetTempDirectory() << "FlushOptOutDropInvalid-" << PAL::getUtcSystemTimeMs() << ".db"; + RemoveDbFiles(dbPath.str()); + config[CFG_STR_CACHE_FILE_PATH] = dbPath.str(); + config[CFG_INT_RAM_QUEUE_SIZE] = 1024 * 1024; + config[CFG_BOOL_ENABLE_BATCHED_STORAGE_FLUSH] = false; + + OfflineStorageHandler handler(logManager, config, dispatcher); + handler.Initialize(observer); + + const size_t kCount = 3; + for (size_t i = 0; i < kCount; i++) + { + StorageRecord r("bad-opt-out-id-" + std::to_string(i), "tenant-token", + EventLatency_Normal, EventPersistence_Normal, /*timestamp*/ 0, + std::vector{ 'x' }); + handler.StoreRecord(r); + } + EXPECT_EQ(handler.GetRecordCount(), kCount); + + handler.Flush(); + + EXPECT_EQ(handler.GetRecordCount(), static_cast(0)); + + handler.Shutdown(); + RemoveDbFiles(dbPath.str()); +} diff --git a/tests/unittests/OfflineStorageTests_SQLite.cpp b/tests/unittests/OfflineStorageTests_SQLite.cpp index 015e197d7..67b843932 100644 --- a/tests/unittests/OfflineStorageTests_SQLite.cpp +++ b/tests/unittests/OfflineStorageTests_SQLite.cpp @@ -162,6 +162,59 @@ TEST_F(OfflineStorageTests_SQLite, GetAndReservedReturnsStoredRecord) EXPECT_THAT(consumer.records[0].reservedUntil, 0); } +TEST_F(OfflineStorageTests_SQLite, StoreRecordsBatchStoresAllRecords) +{ + initializeStorage(); + std::vector batch; + const size_t kCount = 8; + for (size_t i = 0; i < kCount; i++) + { + batch.push_back({ "g" + std::to_string(i), "token", EventLatency_Normal, + EventPersistence_Normal, static_cast(i + 1), { static_cast(i) } }); + } + + // Every record in the batch is stored and individually retrievable. (The + // single-transaction batching is a performance optimization verified by + // benchmarking; this test covers the batch's storage correctness.) + EXPECT_THAT(offlineStorage->StoreRecords(batch), kCount); + + TestRecordConsumer consumer; + EXPECT_THAT(offlineStorage->GetAndReserveRecords(consumer, 100000), true); + ASSERT_THAT(consumer.records.size(), kCount); + for (size_t i = 0; i < kCount; i++) + { + std::string expectedId = "g" + std::to_string(i); + bool found = false; + for (auto const& r : consumer.records) + { + if (r.id == expectedId) { found = true; break; } + } + EXPECT_TRUE(found) << "record " << expectedId << " was not retrieved"; + } +} + +TEST_F(OfflineStorageTests_SQLite, StoreRecordsBatchDropsInvalidAndStoresValid) +{ + initializeStorage(); + std::vector batch = { + { "g1", "token", EventLatency_Normal, EventPersistence_Normal, 1, { 1 } }, // valid + { "g2", "token", EventLatency_Normal, EventPersistence_Normal, 0, { 2 } }, // invalid: timestamp <= 0 + }; + + // The invalid record is reported once during validation. + EXPECT_CALL(observerMock, OnStorageFailed("Invalid parameters")); + + // A permanently-invalid record is dropped (reported once) and the valid + // remainder is still stored. One bad record can never wedge the batch or, via + // a caller that re-queues on a short return (e.g. Flush), block the queue. + EXPECT_THAT(offlineStorage->StoreRecords(batch), static_cast(1)); + + TestRecordConsumer consumer; + EXPECT_THAT(offlineStorage->GetAndReserveRecords(consumer, 100000), true); + ASSERT_THAT(consumer.records.size(), static_cast(1)); + EXPECT_THAT(consumer.records[0].id, "g1"); +} + TEST_F(OfflineStorageTests_SQLite, ReservedRecordIsNotReturned) { initializeStorage(); diff --git a/tests/unittests/PalTests.cpp b/tests/unittests/PalTests.cpp index c931ff376..4f5d49de8 100644 --- a/tests/unittests/PalTests.cpp +++ b/tests/unittests/PalTests.cpp @@ -14,6 +14,8 @@ #include #include #include +#include +#include #ifdef HAVE_MAT_LOGGING #include "pal/PAL.hpp" @@ -225,6 +227,12 @@ namespace void ThrowNonStdException() { throw 123; } void Signal(std::atomic* ran) { ran->store(true); } }; + + class WorkerThreadScheduleTarget + { + public: + void Callback() {} + }; } // A task throwing an exception must be contained by the worker thread loop; @@ -253,6 +261,93 @@ TEST_F(PalTests, WorkerThreadContainsThrowingTask) dispatcher->Join(); } +TEST_F(PalTests, ScheduleTaskAfterWorkerThreadJoinReturnsNoOpHandle) +{ + auto dispatcher = PAL::WorkerThreadFactory::Create(); + dispatcher->Join(); + WorkerThreadScheduleTarget target; + + auto handle = PAL::scheduleTask(dispatcher.get(), 100, &target, &WorkerThreadScheduleTarget::Callback); + + EXPECT_EQ(handle.GetTask(), nullptr); + EXPECT_TRUE(handle.Cancel()); +} + +TEST_F(PalTests, ScheduleTaskHandleClearsAfterWorkerThreadCallbackCompletes) +{ + auto dispatcher = PAL::WorkerThreadFactory::Create(); + std::atomic callbackRan(false); + + class WorkerThreadCompletionTarget + { + public: + explicit WorkerThreadCompletionTarget(std::atomic& callbackRan) : m_callbackRan(callbackRan) {} + void Callback() { m_callbackRan.store(true); } + + private: + std::atomic& m_callbackRan; + } target(callbackRan); + + auto handle = PAL::scheduleTask(dispatcher.get(), 0, &target, &WorkerThreadCompletionTarget::Callback); + + for (int i = 0; i < 500 && !callbackRan.load(); ++i) + PAL::sleep(10); + + ASSERT_TRUE(callbackRan.load()); + EXPECT_EQ(handle.GetTask(), nullptr); + EXPECT_TRUE(handle.Cancel()); + + dispatcher->Join(); +} + +namespace +{ + // Runs on the worker thread and releases the last reference to the dispatcher + // that owns this very thread, exercising the self-dispose path. + class SelfDisposeHelper + { + public: + std::function releaseLastRef; + std::atomic* done = nullptr; + void Run() + { + releaseLastRef(); // drops the last dispatcher reference on its own thread + done->store(true); + } + }; +} + +// The process-wide worker is shared by reference count, and a task can drop the last +// reference from within itself (e.g. by tearing down its LogManager/PAL) while running +// ON the worker thread. The worker must not be freed underneath its own still-running +// threadFunc: it detaches and defers destruction to the thread. This exercises that +// path and must not use-after-free (caught by ASAN). +TEST_F(PalTests, WorkerThreadSelfDisposeOnOwnThreadIsSafe) +{ + auto dispatcher = PAL::WorkerThreadFactory::Create(); + auto* raw = dispatcher.get(); + // 'box' holds the only remaining reference; the task releases it on the worker + // thread. Keep it in a shared box so a copy captured by the task's callable can + // reset it without naming the dispatcher's concrete type. + auto box = std::make_shared(std::move(dispatcher)); + + std::atomic done(false); + SelfDisposeHelper helper; + helper.releaseLastRef = [box]() { box->reset(); }; + helper.done = &done; + + PAL::dispatchTask(raw, &helper, &SelfDisposeHelper::Run); + + for (int i = 0; i < 500 && !done.load(); ++i) + PAL::sleep(10); + ASSERT_TRUE(done.load()); + + // Give the worker time to break its loop and delete itself after the task + // returns. Reaching here without a crash / ASAN report means the object was not + // freed underneath its own threadFunc. + PAL::sleep(200); +} + #ifdef HAVE_MAT_LOGGING class LogInitTest : public Test { diff --git a/tests/unittests/TaskDispatcherCAPITests.cpp b/tests/unittests/TaskDispatcherCAPITests.cpp index b227deb13..4708926da 100644 --- a/tests/unittests/TaskDispatcherCAPITests.cpp +++ b/tests/unittests/TaskDispatcherCAPITests.cpp @@ -229,6 +229,91 @@ TEST(TaskDispatcherCAPITests, Join) EXPECT_EQ(wasJoined, true); } +namespace +{ + // Dispatcher that always drops (and deletes) the task, modeling the + // shutdown-drop path where Queue() cannot report failure. + class DroppingTaskDispatcher : public ITaskDispatcher + { + public: + bool cancelCalled = false; + void Join() override {} + void Queue(MAT::Task* task) override { delete task; } + bool Cancel(MAT::Task* /*task*/, uint64_t /*waitTime*/ = 0) override + { + cancelCalled = true; + return false; + } + }; + + struct NoopCallbackTarget + { + void Callback(int, int) {} + }; +} + +// When the dispatcher drops the task (for example during shutdown), scheduleTask +// must return a no-op handle rather than one pointing at the freed task, so the +// caller never holds a dangling pointer and Cancel() is a safe no-op. +TEST(TaskDispatcherCAPITests, ScheduleTaskReturnsNoOpHandleWhenTaskDropped) +{ + DroppingTaskDispatcher dispatcher; + NoopCallbackTarget target; + + auto handle = scheduleTask(&dispatcher, 100 /*delayMs*/, &target, &NoopCallbackTarget::Callback, 1, 2); + + EXPECT_EQ(handle.GetTask(), nullptr); + EXPECT_TRUE(handle.Cancel()); + EXPECT_FALSE(dispatcher.cancelCalled); +} + +namespace +{ + struct DeferredExecutionState + { + std::string taskId; + task_callback_fn_t callback = nullptr; + bool cancelCalled = false; + }; + + static std::unique_ptr s_deferredExecutionState; + + void EVTSDK_LIBABI_CDECL OnDeferredTaskDispatcherQueue(evt_task_t* task, task_callback_fn_t callback) + { + s_deferredExecutionState->taskId = task->id; + s_deferredExecutionState->callback = callback; + } + + bool EVTSDK_LIBABI_CDECL OnDeferredTaskDispatcherCancel(const char* taskId) + { + s_deferredExecutionState->cancelCalled = true; + return (s_deferredExecutionState->taskId == taskId); + } + + void EVTSDK_LIBABI_CDECL OnDeferredTaskDispatcherJoin() + {} +} + +TEST(TaskDispatcherCAPITests, ScheduleTaskHandleClearsAfterAsyncCallbackCompletes) +{ + TaskDispatcher_CAPI taskDispatcher(&OnDeferredTaskDispatcherQueue, &OnDeferredTaskDispatcherCancel, &OnDeferredTaskDispatcherJoin); + s_deferredExecutionState.reset(new DeferredExecutionState()); + + NoopCallbackTarget target; + auto handle = scheduleTask(&taskDispatcher, 100 /*delayMs*/, &target, &NoopCallbackTarget::Callback, 1, 2); + + ASSERT_NE(handle.GetTask(), nullptr); + ASSERT_NE(s_deferredExecutionState->callback, nullptr); + + s_deferredExecutionState->callback(s_deferredExecutionState->taskId.c_str()); + + EXPECT_EQ(handle.GetTask(), nullptr); + EXPECT_TRUE(handle.Cancel()); + EXPECT_FALSE(s_deferredExecutionState->cancelCalled); + + s_deferredExecutionState.reset(); +} + TEST(TaskDispatcherCAPITests, ExecuteCallbackThatThrowsIsContained) { TaskDispatcher_CAPI taskDispatcher(&OnTaskDispatcherQueue, &OnTaskDispatcherCancel, &OnTaskDispatcherJoin); @@ -247,4 +332,3 @@ TEST(TaskDispatcherCAPITests, ExecuteCallbackThatThrowsIsContained) EXPECT_NO_THROW(dispatchTask(&taskDispatcher, testHelper.get(), &TestHelper::Callback, 10 /*param1*/, 20 /*param2*/)); EXPECT_EQ(wasExecuted, true); } - diff --git a/tests/unittests/TransmissionPolicyManagerTests.cpp b/tests/unittests/TransmissionPolicyManagerTests.cpp index 6cbdb99f5..2f4a75ec1 100644 --- a/tests/unittests/TransmissionPolicyManagerTests.cpp +++ b/tests/unittests/TransmissionPolicyManagerTests.cpp @@ -11,14 +11,24 @@ #include "tpm/TransmissionPolicyManager.hpp" #include "TransmitProfiles.hpp" +#include +#include +#include +#include + using namespace testing; using namespace MAT; class TransmissionPolicyManager4Test : public TransmissionPolicyManager { public: + TransmissionPolicyManager4Test(ITelemetrySystem& system, ITaskDispatcher& taskDispatcher, IBandwidthController* bandwidthController) + : TransmissionPolicyManager(system, taskDispatcher, bandwidthController) + { + } + TransmissionPolicyManager4Test(ITelemetrySystem& system, IBandwidthController* bandwidthController) - : TransmissionPolicyManager(system, *PAL::getDefaultTaskDispatcher(), bandwidthController) + : TransmissionPolicyManager4Test(system, *PAL::getDefaultTaskDispatcher(), bandwidthController) { } @@ -69,6 +79,141 @@ class TransmissionPolicyManager4Test : public TransmissionPolicyManager { } }; +class BlockingCancelTaskDispatcher : public ITaskDispatcher +{ +public: + ~BlockingCancelTaskDispatcher() override + { + Join(); + } + + void Join() override + { + std::lock_guard lock(m_tasksMutex); + for (auto* task : m_tasks) + { + delete task; + } + m_tasks.clear(); + } + + void Queue(Task* task) override + { + std::lock_guard lock(m_tasksMutex); + m_tasks.push_back(task); + } + + bool Cancel(Task* task, uint64_t waitTime = 0) override + { + UNREFERENCED_PARAMETER(waitTime); + + { + std::lock_guard lock(m_tasksMutex); + auto it = std::find(m_tasks.begin(), m_tasks.end(), task); + if (it == m_tasks.end()) + { + return false; + } + delete *it; + m_tasks.erase(it); + } + + { + std::lock_guard lock(m_cancelMutex); + m_cancelEntered = true; + } + m_cancelEnteredCv.notify_all(); + + std::unique_lock lock(m_cancelMutex); + m_cancelReleasedCv.wait(lock, [this]() { return m_cancelReleased; }); + return true; + } + + bool WaitForCancel(const std::chrono::milliseconds timeout) + { + std::unique_lock lock(m_cancelMutex); + return m_cancelEnteredCv.wait_for(lock, timeout, [this]() { return m_cancelEntered; }); + } + + void ReleaseCancel() + { + { + std::lock_guard lock(m_cancelMutex); + m_cancelReleased = true; + } + m_cancelReleasedCv.notify_all(); + } + +private: + std::mutex m_tasksMutex; + std::vector m_tasks; + + std::mutex m_cancelMutex; + std::condition_variable m_cancelEnteredCv; + std::condition_variable m_cancelReleasedCv; + bool m_cancelEntered = false; + bool m_cancelReleased = false; +}; + +class RunningTaskDispatcher : public ITaskDispatcher +{ +public: + ~RunningTaskDispatcher() override + { + std::lock_guard lock(m_tasksMutex); + for (auto* task : m_tasks) + { + delete task; + } + m_tasks.clear(); + } + + void Join() override + { + std::lock_guard lock(m_tasksMutex); + for (auto* task : m_tasks) + { + delete task; + } + m_tasks.clear(); + } + + void Queue(Task* task) override + { + std::lock_guard lock(m_tasksMutex); + m_tasks.push_back(task); + } + + bool Cancel(Task* task, uint64_t waitTime = 0) override + { + UNREFERENCED_PARAMETER(task); + UNREFERENCED_PARAMETER(waitTime); + // Simulate a task that is currently executing on the worker: + // cancellation can never proceed without waiting for the run + // to complete, so a no-wait cancel must return false. + std::lock_guard lock(m_tasksMutex); + m_cancelCount++; + return false; + } + + size_t QueuedCount() const + { + std::lock_guard lock(m_tasksMutex); + return m_tasks.size(); + } + + size_t CancelCount() const + { + std::lock_guard lock(m_tasksMutex); + return m_cancelCount; + } + +private: + mutable std::mutex m_tasksMutex; + std::vector m_tasks; + size_t m_cancelCount = 0; +}; + class TransmissionPolicyManagerTests : public StrictMock { protected: StrictMock runtimeConfigMock; @@ -608,6 +753,73 @@ TEST_F(TransmissionPolicyManagerTests, cancelUploadTask_ScheduledUpload_IsUpload ASSERT_FALSE(tpm.m_isUploadScheduled); } +TEST_F(TransmissionPolicyManagerTests, ForceScheduleRetainsImmediateUploadWhenCancelBlocks) +{ + BlockingCancelTaskDispatcher dispatcher; + TransmissionPolicyManager4Test blockingTpm(testing::getSystem(), dispatcher, &bandwidthControllerMock); + blockingTpm.paused(false); + + blockingTpm.scheduleUploadParent(std::chrono::milliseconds{ 1000 }, EventLatency_Normal, false); + auto delayedUploadTime = blockingTpm.m_scheduledUploadTime; + + auto forceSchedule = std::async(std::launch::async, [&blockingTpm]() { + blockingTpm.scheduleUploadParent(std::chrono::milliseconds{}, EventLatency_RealTime, true); + }); + + if (!dispatcher.WaitForCancel(std::chrono::milliseconds{ 250 })) + { + dispatcher.ReleaseCancel(); + forceSchedule.get(); + FAIL() << "Timed out waiting for cancel to block"; + } + + auto delayedSchedule = std::async(std::launch::async, [&blockingTpm]() { + blockingTpm.scheduleUploadParent(std::chrono::milliseconds{ 1000 }, EventLatency_Normal, false); + }); + + EXPECT_EQ(delayedSchedule.wait_for(std::chrono::milliseconds{ 100 }), std::future_status::timeout); + + dispatcher.ReleaseCancel(); + + forceSchedule.get(); + delayedSchedule.get(); + + ASSERT_TRUE(blockingTpm.m_isUploadScheduled); + EXPECT_LT(blockingTpm.m_scheduledUploadTime, delayedUploadTime); +} + +TEST_F(TransmissionPolicyManagerTests, ForceScheduleAppliesLatencyWhenRunningCancelFails) +{ + RunningTaskDispatcher dispatcher; + TransmissionPolicyManager4Test runningTpm(testing::getSystem(), dispatcher, &bandwidthControllerMock); + runningTpm.paused(false); + + // Queue an initial upload so m_scheduledUpload has a non-null task and + // m_isUploadScheduled is set; the dispatcher's Cancel will fail later + // (simulating the "task currently executing on worker" race). + runningTpm.scheduleUploadParent(std::chrono::milliseconds{ 1000 }, EventLatency_Normal, false); + ASSERT_TRUE(runningTpm.m_isUploadScheduled); + ASSERT_EQ(dispatcher.QueuedCount(), 1u); + + auto scheduledTimeBefore = runningTpm.m_scheduledUploadTime; + // Reset m_runningLatency so we can observe the force path updating it + // (the initial schedule may have bumped it depending on the active + // profile's timers). + runningTpm.runningLatency(EventLatency_Normal); + + // Force a higher-priority schedule. The dispatcher's no-wait cancel + // returns false, so the previous task remains in flight. The fix in + // scheduleUpload must propagate the new latency to m_runningLatency + // so the running task picks it up under the same mutex. + runningTpm.scheduleUploadParent(std::chrono::milliseconds{}, EventLatency_RealTime, true); + + EXPECT_GE(dispatcher.CancelCount(), 1u); + EXPECT_EQ(dispatcher.QueuedCount(), 1u); + EXPECT_TRUE(runningTpm.m_isUploadScheduled); + EXPECT_EQ(runningTpm.m_runningLatency, EventLatency_RealTime); + EXPECT_EQ(runningTpm.m_scheduledUploadTime, scheduledTimeBefore); +} + TEST_F(TransmissionPolicyManagerTests, increaseBackoff_EmptyBackoffObject_ReturnZero) { tpm.m_backoff = nullptr; diff --git a/tests/unittests/TransmitProfilesTests.cpp b/tests/unittests/TransmitProfilesTests.cpp index 58e9d36b5..ce8839de5 100644 --- a/tests/unittests/TransmitProfilesTests.cpp +++ b/tests/unittests/TransmitProfilesTests.cpp @@ -375,6 +375,24 @@ R"([{ ASSERT_TRUE(TransmitProfiles::load(badRule)); } +TEST_F(TransmitProfilesTests, load_Json_RuleWithLowBatteryPowerState_MapsToPowerSourceLowBattery) +{ + // A rule using the "low_battery" powerState must map to PowerSource_LowBattery + // rather than silently falling back to the default PowerSource_Any. + const std::string profile = +R"([{ + "name": "LowBatteryProfile", + "rules": [ + { "powerState": "low_battery", "timers": [ 8, 4, 2 ] } + ] +}])"; + + ASSERT_TRUE(TransmitProfiles::load(profile)); + const auto& rules = TransmitProfiles::profiles[std::string{"LowBatteryProfile"}].rules; + ASSERT_EQ(rules.size(), size_t{1}); + ASSERT_EQ(rules[0].powerState, PowerSource_LowBattery); +} + /* The following tests probably should not pass. But they do.