Fix runtime data races, memory leak, and shutdown safety

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

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);
         }
     }

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;

lib/include/public/ITaskDispatcher.hpp

@@ -121,11 +121,11 @@ namespace MAT_NS_BEGIN
         /// <param name="waitTime">Amount of time to wait for if the task is currently executing</param>
         /// <returns>True if successfully cancelled, else false</returns>
         virtual bool Cancel(Task* task, uint64_t waitTime = 0) = 0;
+
     };
 
     /// @endcond
 
 } MAT_NS_END
 
 #endif // ITASKDISPATCHER_HPP
-

lib/offline/OfflineStorageHandler.cpp

@@ -164,6 +164,14 @@ namespace MAT_NS_BEGIN {
     void OfflineStorageHandler::Flush()
     {
         if (!m_logManager.StartActivity()) {
+            // 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

lib/pal/TaskDispatcher.hpp

@@ -16,6 +16,7 @@
 #include <climits>
 #include <algorithm>
 #include <atomic>
+#include <memory>
 #include <utility>
 
 #include "ITaskDispatcher.hpp"
@@ -25,6 +26,15 @@ namespace PAL_NS_BEGIN {
 
     namespace detail {
 
+        struct TaskLifetimeState
+        {
+            TaskLifetimeState() :
+                task(nullptr)
+            {}
+
+            std::atomic<MAT::Task*> task;
+        };
+
         template<typename TCall>
         class TaskCall : public Task
         {
@@ -48,14 +58,35 @@ namespace PAL_NS_BEGIN {
                 this->TargetTime = targetTime;
             }
 
+            TaskCall(TCall& call, int64_t targetTime, std::shared_ptr<TaskLifetimeState> 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<TaskLifetimeState> m_lifetimeState;
         };
 
     } // namespace detail
@@ -121,9 +152,17 @@ 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<TPassedArgs>(args)...);
-        auto task = new detail::TaskCall<decltype(bound)>(bound, getMonotonicTimeMs() + (int64_t)delayMs);
+        auto taskLifetime = std::make_shared<detail::TaskLifetimeState>();
+        auto task = new detail::TaskCall<decltype(bound)>(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.
+        auto queuedTask = taskLifetime->task.load(std::memory_order_acquire);
+        if (queuedTask == nullptr)
+        {
+            return DeferredCallbackHandle();
+        }
+        return DeferredCallbackHandle(queuedTask, taskDispatcher);
     }
 
     template<typename TObject, typename... TFuncArgs, typename... TPassedArgs>
@@ -135,4 +174,3 @@ namespace PAL_NS_BEGIN {
 } PAL_NS_END
 
 #endif
-

lib/pal/WorkerThread.cpp

@@ -7,6 +7,8 @@
 #include "pal/PAL.hpp"
 
 #include <exception>
+#include <system_error>
+#include <atomic>
 
 #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<std::thread::id>,
+        // 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;
@@ -37,15 +47,19 @@ namespace PAL_NS_BEGIN {
         std::list<MAT::Task*> m_timerQueue;
         Event                 m_event;
         MAT::Task*            m_itemInProgress;
-        int count = 0;
+        bool                  m_shuttingDown = false;
+        // 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<bool>     m_disposeFromThread { false };
 
     public:
 
         WorkerThread()
         {
             m_itemInProgress = nullptr;
             m_hThread = std::thread(WorkerThread::threadFunc, static_cast<void*>(this));
-            LOG_INFO("Started new thread %u", m_hThread.get_id());
+            LOG_INFO("Started new thread %zu", std::hash<std::thread::id>{}(m_hThread.get_id()));
         }
 
         ~WorkerThread()
@@ -55,32 +69,99 @@ namespace PAL_NS_BEGIN {
 
         void Join() final
         {
-            auto item = new WorkerThreadShutdownItem();
-            Queue(item);
             std::thread::id this_id = std::this_thread::get_id();
+            bool joined = false;
+            {
+                LOCKGUARD(m_lock);
+                if (!m_shuttingDown) {
+                    m_shuttingDown = true;
+                    m_queue.push_back(new WorkerThreadShutdownItem());
+                    m_event.post();
+                }
+            }
             try {
-                if (m_hThread.joinable() && (m_hThread.get_id() != this_id))
+                if (!m_hThread.joinable()) {
+                    return;
+                }
+                if (m_hThread.get_id() != this_id) {
                     m_hThread.join();
-                else
+                    joined = true;
+                } else {
                     m_hThread.detach();
+                }
+            }
+            catch (const std::system_error& e) {
+                LOG_ERROR("Thread join/detach failed: [%d] %s", e.code().value(), e.what());
+            }
+            catch (const std::exception& e) {
+                LOG_ERROR("Thread join/detach failed: %s", e.what());
             }
-            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()) {
+                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) {
+                for (auto task : m_queue) { delete task; }
+                m_queue.clear();
+                for (auto task : m_timerQueue) { delete task; }
+                m_timerQueue.clear();
             }
-            if (!m_timerQueue.empty())
+        }
+
+        // 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!");
+                if (!m_shuttingDown) {
+                    m_shuttingDown = true;
+                    m_queue.push_back(new WorkerThreadShutdownItem());
+                    m_event.post();
+                }
+                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<void*>(item));
             LOCKGUARD(m_lock);
+            if (m_shuttingDown) {
+                LOG_WARN("Dropping queued task %p during shutdown", static_cast<void*>(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 +172,6 @@ namespace PAL_NS_BEGIN {
             else {
                 m_queue.push_back(item);
             }
-            count++;
             m_event.post();
         }
 
@@ -181,7 +261,11 @@ namespace PAL_NS_BEGIN {
             uint64_t wakeupCount = 0;
 
             WorkerThread* self = reinterpret_cast<WorkerThread*>(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::thread::id>{}(std::this_thread::get_id()));
 
             for (;;) {
                 std::unique_ptr<MAT::Task> item = nullptr;
@@ -231,6 +315,19 @@ namespace PAL_NS_BEGIN {
                 if (item->Type == MAT::Task::Shutdown) {
                     item.reset();
                     self->m_itemInProgress = nullptr;
+                    // 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);
+                        for (auto task : self->m_queue) { delete task; }
+                        self->m_queue.clear();
+                        for (auto task : self->m_timerQueue) { delete task; }
+                        self->m_timerQueue.clear();
+                    }
                     break;
                 }
 
@@ -262,17 +359,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<ITaskDispatcher> Create()
         {
-            return std::make_shared<WorkerThread>();
+            // 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<WorkerThread>(
+                new WorkerThread(),
+                [](WorkerThread* self) { if (self->onLastReferenceReleased()) delete self; });
         }
     }
 
 } PAL_NS_END
 
 #endif
-