Replace SubscriptionData value map with shared_ptr; drop stale notifi…

src/gateway.cpp

@@ -22,6 +22,7 @@
 #include "firebolt/types.h"
 #include "transport.h"
 #include "utils.h"
+#include <any>
 #include <assert.h>
 #include <chrono>
 #include <condition_variable>
@@ -361,7 +362,27 @@ class Server
 
             for (auto& callback : notification.callbacks)
             {
-                callback.lambda(callback.usercb, notification.params);
+                try
+                {
+                    callback.lambda(callback.usercb, notification.params);
+                }
+                catch (const std::bad_any_cast& e)
+                {
+                    FIREBOLT_LOG_ERROR("Gateway",
+                                       "[notification-worker] bad_any_cast dispatching event='%s': %s — "
+                                       "notification type does not match the registered callback signature",
+                                       callback.eventName.c_str(), e.what());
+                }
+                catch (const std::exception& e)
+                {
+                    FIREBOLT_LOG_ERROR("Gateway", "[notification-worker] exception dispatching event='%s': %s",
+                                       callback.eventName.c_str(), e.what());
+                }
+                catch (...)
+                {
+                    FIREBOLT_LOG_ERROR("Gateway", "[notification-worker] unknown exception dispatching event='%s'",
+                                       callback.eventName.c_str());
+                }
             }
         }
     }

src/helpers_impl.h

@@ -18,6 +18,9 @@
 
 #include "firebolt/gateway.h"
 #include "firebolt/helpers.h"
+#include <map>
+#include <memory>
+#include <mutex>
 
 namespace Firebolt::Helpers
 {
@@ -35,8 +38,8 @@ class HelperImpl : public IHelper
         std::lock_guard<std::mutex> lock(mutex_);
         for (auto& subscription : subscriptions_)
         {
-            void* notificationPtr = reinterpret_cast<void*>(&subscription.second);
-            gateway_.unsubscribe(subscription.second.eventName, notificationPtr);
+            void* notificationPtr = static_cast<void*>(subscription.second.get());
+            gateway_.unsubscribe(subscription.second->eventName, notificationPtr);
         }
         subscriptions_.clear();
     }
@@ -76,8 +79,8 @@ class HelperImpl : public IHelper
         {
             return Result<void>{Error::General};
         }
-        void* notificationPtr = reinterpret_cast<void*>(&it->second);
-        auto errorStatus{gateway_.unsubscribe(it->second.eventName, notificationPtr)};
+        void* notificationPtr = static_cast<void*>(it->second.get());
+        auto errorStatus{gateway_.unsubscribe(it->second->eventName, notificationPtr)};
         subscriptions_.erase(it);
         return Result<void>{errorStatus};
     }
@@ -87,10 +90,10 @@ class HelperImpl : public IHelper
         std::lock_guard<std::mutex> lock(mutex_);
         for (auto it = subscriptions_.begin(); it != subscriptions_.end();)
         {
-            if (it->second.owner == owner)
+            if (it->second->owner == owner)
             {
-                void* notificationPtr = reinterpret_cast<void*>(&it->second);
-                gateway_.unsubscribe(it->second.eventName, notificationPtr);
+                void* notificationPtr = static_cast<void*>(it->second.get());
+                gateway_.unsubscribe(it->second->eventName, notificationPtr);
                 it = subscriptions_.erase(it);
             }
             else
@@ -116,10 +119,31 @@ class HelperImpl : public IHelper
     {
         std::lock_guard<std::mutex> lock(mutex_);
         uint64_t newId = currentId_++;
-        subscriptions_[newId] = SubscriptionData{owner, eventName, std::move(notification)};
-        void* notificationPtr = reinterpret_cast<void*>(&subscriptions_[newId]);
+        auto spData = std::make_shared<SubscriptionData>(SubscriptionData{owner, eventName, std::move(notification)});
+        subscriptions_[newId] = spData;
+        void* notificationPtr = static_cast<void*>(spData.get());
 
-        Error status = gateway_.subscribe(eventName, callback, notificationPtr);
+        // Guard the callback with a weak_ptr so that any notification already queued
+        // to the async worker thread at the time of unsubscribe is safely dropped
+        // rather than invoking the callback through a dangling pointer.  This closes
+        // the race between Server::notify() copying callbacks under eventMap_mtx and
+        // the worker dispatching them after SubscriptionData has been destroyed.
+        std::weak_ptr<SubscriptionData> wpData = spData;
+        Firebolt::Transport::EventCallback wrappedCallback =
+            [wpData, callback, eventName](void* /*usercb*/, const nlohmann::json& json)
+        {
+            if (auto sp = wpData.lock())
+            {
+                callback(sp.get(), json);
+            }
+            else
+            {
+                FIREBOLT_LOG_DEBUG("Helper", "[subscription] notification dropped for already-unsubscribed event='%s'",
+                                   eventName.c_str());
+            }
+        };
+
+        Error status = gateway_.subscribe(eventName, std::move(wrappedCallback), notificationPtr);
 
         if (Error::None != status)
         {
@@ -131,7 +155,7 @@ class HelperImpl : public IHelper
 
     Firebolt::Transport::IGateway& gateway_;
     std::mutex mutex_;
-    std::map<uint64_t, SubscriptionData> subscriptions_;
+    std::map<uint64_t, std::shared_ptr<SubscriptionData>> subscriptions_;
     uint64_t currentId_{0};
 };
 } // namespace Firebolt::Helpers

test/unit/gatewayTest.cpp

@@ -24,6 +24,7 @@
 #include <gtest/gtest.h>
 #include <netinet/in.h>
 #include <nlohmann/json.hpp>
+#include <stdexcept>
 #include <sys/socket.h>
 #include <thread>
 #include <unistd.h>
@@ -1478,6 +1479,55 @@ TEST_F(GatewayUTest, LegacyUnsubscribeIteratesPastNonMatchingEntry)
     EXPECT_EQ(err, Firebolt::Error::None);
 }
 
+// ---------------------------------------------------------------------------
+// Test name: GatewayUTest.NotificationWorkerContinuesAfterCallbackException
+// Covers: src/gateway.cpp: notification worker for-loop catch block — the
+//         worker must continue dispatching remaining callbacks in the same
+//         notification batch even when one callback throws std::exception.
+//         Regression: before the try/catch was added, the exception would
+//         escape the thread entry point and call std::terminate.
+// Scenario type: regression
+// ---------------------------------------------------------------------------
+TEST_F(GatewayUTest, NotificationWorkerContinuesAfterCallbackException)
+{
+    IGateway& gateway = connectAndWait();
+
+    // Callback A intentionally throws — exercises catch(const std::exception&)
+    auto onEventA = [](void*, const nlohmann::json&) { throw std::runtime_error("test: simulated callback exception"); };
+    int cbA = 0;
+
+    // Callback B signals via a promise — confirms the worker reached it after A threw
+    std::promise<bool> deliveredPromise;
+    auto deliveredFuture = deliveredPromise.get_future();
+    auto onEventB = [](void* usercb, const nlohmann::json&)
+    { static_cast<std::promise<bool>*>(usercb)->set_value(true); };
+
+    // Both subscribe to the same event so they land in the same notification.callbacks vector
+    Firebolt::Error err = gateway.subscribe("test.onWorkerContinues", onEventA, &cbA);
+    EXPECT_EQ(err, Firebolt::Error::None);
+    err = gateway.subscribe("test.onWorkerContinues", onEventB, &deliveredPromise);
+    EXPECT_EQ(err, Firebolt::Error::None);
+
+    // Server fires the event once
+    m_onMessageAction = [](server* s, connection_hdl hdl)
+    {
+        nlohmann::json eventMsg;
+        eventMsg["jsonrpc"] = "2.0";
+        eventMsg["method"] = "test.onWorkerContinues";
+        eventMsg["params"] = {{"fired", true}};
+        s->send(hdl, eventMsg.dump(), websocketpp::frame::opcode::text);
+    };
+    gateway.send("dummy.message", {});
+
+    // Callback B must still fire — worker continued the loop despite A throwing
+    auto status = deliveredFuture.wait_for(std::chrono::seconds(2));
+    ASSERT_EQ(status, std::future_status::ready) << "Notification worker stopped after exception in callback A";
+    EXPECT_TRUE(deliveredFuture.get());
+
+    gateway.unsubscribe("test.onWorkerContinues", &cbA);
+    gateway.unsubscribe("test.onWorkerContinues", &deliveredPromise);
+}
+
 // Regression test: disconnect() must cancel all pending requests so that calling
 // threads blocked on future.get() unblock immediately with NotConnected rather
 // than hanging indefinitely.

test/unit/helperTest.cpp

@@ -605,3 +605,135 @@ TEST(GetHelperInstanceTest, ReturnsSingleton)
     // Same singleton reference
     EXPECT_EQ(&helper1, &helper2);
 }
+
+// ---------------------------------------------------------------------------
+// Race condition fix tests (MONUI-908)
+//
+// These tests validate the weak_ptr guard introduced in HelperImpl::subscribe.
+// The guard prevents a use-after-free crash that occurred when:
+//   1. The platform sent an event notification (e.g. onSpeechInterrupted).
+//   2. Before the worker thread dispatched it, unsubscribeAll() ran and
+//      destroyed the SubscriptionData object.
+//   3. The worker thread then invoked the registered callback with a dangling
+//      void* pointer → bad_any_cast → std::terminate → device crash.
+TEST_F(HelperUTest, WeakPtrGuard_NotificationDeliveredWhileSubscribed)
+{
+    Firebolt::Transport::EventCallback capturedCallback;
+    void* capturedUsercb = nullptr;
+
+    EXPECT_CALL(mockGateway, subscribe("test.onEvent", _, _))
+        .WillOnce(
+            [&](const std::string&, Firebolt::Transport::EventCallback cb, void* ucb)
+            {
+                capturedCallback = std::move(cb);
+                capturedUsercb = ucb;
+                return Error::None;
+            });
+
+    std::promise<int> deliveredValue;
+    auto future = deliveredValue.get_future();
+    std::function<void(int)> notification = [&deliveredValue](int val) { deliveredValue.set_value(val); };
+
+    IHelper& ihelper = helper;
+    auto subResult =
+        ihelper.subscribe(this, "test.onEvent", std::move(notification), onPropertyChangedCallback<TestJson, int>);
+    ASSERT_TRUE(subResult);
+    ASSERT_TRUE(capturedCallback) << "wrappedCallback must have been captured";
+
+    // Worker thread dispatches notification while subscription is still alive
+    capturedCallback(capturedUsercb, nlohmann::json{{"value", 42}});
+
+    auto status = future.wait_for(std::chrono::seconds(1));
+    ASSERT_EQ(status, std::future_status::ready);
+    EXPECT_EQ(future.get(), 42);
+
+    // Destructor will unsubscribe the remaining active subscription
+    EXPECT_CALL(mockGateway, unsubscribe("test.onEvent", _)).WillOnce(Return(Error::None));
+}
+
+// ---------------------------------------------------------------------------
+// Test name: HelperUTest.WeakPtrGuard_NotificationDroppedAfterUnsubscribeAll
+// Covers: wrappedCallback drops in-flight notification when SubscriptionData
+//         has been destroyed by unsubscribeAll() — the exact MONUI-908 race.
+//
+//   Before fix: capturedUsercb is dangling → bad_any_cast → std::terminate
+//   After fix:  wpData.lock() returns nullptr → silently dropped, no crash
+//
+// Scenario type: race condition regression test
+// ---------------------------------------------------------------------------
+TEST_F(HelperUTest, WeakPtrGuard_NotificationDroppedAfterUnsubscribeAll)
+{
+    Firebolt::Transport::EventCallback capturedCallback;
+    void* capturedUsercb = nullptr;
+
+    EXPECT_CALL(mockGateway, subscribe("tts.onSpeechInterrupted", _, _))
+        .WillOnce(
+            [&](const std::string&, Firebolt::Transport::EventCallback cb, void* ucb)
+            {
+                capturedCallback = std::move(cb);
+                capturedUsercb = ucb;
+                return Error::None;
+            });
+
+    bool notificationFired = false;
+    std::function<void(int)> notification = [&notificationFired](int) { notificationFired = true; };
+
+    IHelper& ihelper = helper;
+    auto subResult = ihelper.subscribe(this, "tts.onSpeechInterrupted", std::move(notification),
+                                       onPropertyChangedCallback<TestJson, int>);
+    ASSERT_TRUE(subResult);
+
+    // Lifecycle background event fires → Monarch calls unsubscribeAll()
+    // shared_ptr ref count drops to 0 → SubscriptionData is destroyed
+    EXPECT_CALL(mockGateway, unsubscribe("tts.onSpeechInterrupted", _)).WillOnce(Return(Error::None));
+    helper.unsubscribeAll(this);
+
+    // Platform sends onSpeechInterrupted — notification was already queued in-flight
+    // Replay the race: call the wrappedCallback with the now-stale usercb address.
+    ASSERT_TRUE(capturedCallback) << "wrappedCallback must have been captured";
+    capturedCallback(capturedUsercb, nlohmann::json{{"value", 1}});
+
+    // Notification must NOT be delivered to Monarch
+    EXPECT_FALSE(notificationFired);
+    // Reaching this line without crashing IS the primary assertion of this test
+}
+
+// ---------------------------------------------------------------------------
+// Test name: HelperUTest.WeakPtrGuard_NotificationDroppedAfterUnsubscribeById
+// Covers: Same weak_ptr guard works for single unsubscribe (not just unsubscribeAll)
+// Scenario type: race condition regression test
+// ---------------------------------------------------------------------------
+TEST_F(HelperUTest, WeakPtrGuard_NotificationDroppedAfterUnsubscribeById)
+{
+    Firebolt::Transport::EventCallback capturedCallback;
+    void* capturedUsercb = nullptr;
+
+    EXPECT_CALL(mockGateway, subscribe("test.onEvent", _, _))
+        .WillOnce(
+            [&](const std::string&, Firebolt::Transport::EventCallback cb, void* ucb)
+            {
+                capturedCallback = std::move(cb);
+                capturedUsercb = ucb;
+                return Error::None;
+            });
+
+    bool notificationFired = false;
+    std::function<void(int)> notification = [&notificationFired](int) { notificationFired = true; };
+
+    IHelper& ihelper = helper;
+    auto subResult =
+        ihelper.subscribe(this, "test.onEvent", std::move(notification), onPropertyChangedCallback<TestJson, int>);
+    ASSERT_TRUE(subResult);
+    SubscriptionId id = *subResult;
+
+    // Unsubscribe by specific ID → SubscriptionData destroyed
+    EXPECT_CALL(mockGateway, unsubscribe("test.onEvent", _)).WillOnce(Return(Error::None));
+    helper.unsubscribe(id);
+
+    // In-flight notification arrives after the subscription was removed
+    ASSERT_TRUE(capturedCallback) << "wrappedCallback must have been captured";
+    capturedCallback(capturedUsercb, nlohmann::json{{"value", 99}});
+
+    EXPECT_FALSE(notificationFired);
+    // Reaching this line without crashing IS the primary assertion of this test
+}