diff --git a/include/firebolt/helpers.h b/include/firebolt/helpers.h index 0910b66..f4a1b44 100644 --- a/include/firebolt/helpers.h +++ b/include/firebolt/helpers.h @@ -40,10 +40,10 @@ template void onPropertyChangedCallback(void* subscriptionDataPtr, const nlohmann::json& jsonResponse) { SubscriptionData* subscriptionData = reinterpret_cast(subscriptionDataPtr); - auto notifier = std::any_cast>(subscriptionData->notification); - JsonType jsonType; try { + auto notifier = std::any_cast>(subscriptionData->notification); + JsonType jsonType; jsonType.fromJson(jsonResponse); if constexpr (sizeof...(Args) > 1) { @@ -54,9 +54,14 @@ void onPropertyChangedCallback(void* subscriptionDataPtr, const nlohmann::json& notifier(jsonType.value()); } } + catch (const std::bad_any_cast& e) + { + FIREBOLT_LOG_ERROR("Event", "Notification type mismatch for event '%s': %s", + subscriptionData->eventName.c_str(), e.what()); + } catch (const std::exception& e) { - FIREBOLT_LOG_ERROR("Event", "Cannot parse event data for event %s, payload: %s", + FIREBOLT_LOG_ERROR("Event", "Cannot parse event data for event '%s', payload: %s", subscriptionData->eventName.c_str(), jsonResponse.dump().c_str()); } } diff --git a/src/gateway.cpp b/src/gateway.cpp index 9a1e028..3017ce6 100644 --- a/src/gateway.cpp +++ b/src/gateway.cpp @@ -22,6 +22,7 @@ #include "firebolt/types.h" #include "transport.h" #include "utils.h" +#include #include #include #include @@ -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()); + } } } } diff --git a/src/helpers_impl.h b/src/helpers_impl.h index 82b1e5b..bb7fabf 100644 --- a/src/helpers_impl.h +++ b/src/helpers_impl.h @@ -18,6 +18,9 @@ #include "firebolt/gateway.h" #include "firebolt/helpers.h" +#include +#include +#include namespace Firebolt::Helpers { @@ -35,8 +38,8 @@ class HelperImpl : public IHelper std::lock_guard lock(mutex_); for (auto& subscription : subscriptions_) { - void* notificationPtr = reinterpret_cast(&subscription.second); - gateway_.unsubscribe(subscription.second.eventName, notificationPtr); + void* notificationPtr = static_cast(subscription.second.get()); + gateway_.unsubscribe(subscription.second->eventName, notificationPtr); } subscriptions_.clear(); } @@ -76,8 +79,8 @@ class HelperImpl : public IHelper { return Result{Error::General}; } - void* notificationPtr = reinterpret_cast(&it->second); - auto errorStatus{gateway_.unsubscribe(it->second.eventName, notificationPtr)}; + void* notificationPtr = static_cast(it->second.get()); + auto errorStatus{gateway_.unsubscribe(it->second->eventName, notificationPtr)}; subscriptions_.erase(it); return Result{errorStatus}; } @@ -87,10 +90,10 @@ class HelperImpl : public IHelper std::lock_guard lock(mutex_); for (auto it = subscriptions_.begin(); it != subscriptions_.end();) { - if (it->second.owner == owner) + if (it->second->owner == owner) { - void* notificationPtr = reinterpret_cast(&it->second); - gateway_.unsubscribe(it->second.eventName, notificationPtr); + void* notificationPtr = static_cast(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 lock(mutex_); uint64_t newId = currentId_++; - subscriptions_[newId] = SubscriptionData{owner, eventName, std::move(notification)}; - void* notificationPtr = reinterpret_cast(&subscriptions_[newId]); + auto spData = std::make_shared(SubscriptionData{owner, eventName, std::move(notification)}); + subscriptions_[newId] = spData; + void* notificationPtr = static_cast(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 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 subscriptions_; + std::map> subscriptions_; uint64_t currentId_{0}; }; } // namespace Firebolt::Helpers diff --git a/test/unit/gatewayTest.cpp b/test/unit/gatewayTest.cpp index 693117c..67b519e 100644 --- a/test/unit/gatewayTest.cpp +++ b/test/unit/gatewayTest.cpp @@ -24,6 +24,7 @@ #include #include #include +#include #include #include #include @@ -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 deliveredPromise; + auto deliveredFuture = deliveredPromise.get_future(); + auto onEventB = [](void* usercb, const nlohmann::json&) + { static_cast*>(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. diff --git a/test/unit/helperTest.cpp b/test/unit/helperTest.cpp index be93df3..73792cb 100644 --- a/test/unit/helperTest.cpp +++ b/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 deliveredValue; + auto future = deliveredValue.get_future(); + std::function notification = [&deliveredValue](int val) { deliveredValue.set_value(val); }; + + IHelper& ihelper = helper; + auto subResult = + ihelper.subscribe(this, "test.onEvent", std::move(notification), onPropertyChangedCallback); + ASSERT_TRUE(subResult); + ASSERT_TRUE(capturedCallback) << "wrappedCallback must have been captured"; + + // Simulate the worker thread dispatching a notification while the 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 notification = [¬ificationFired](int) { notificationFired = true; }; + + IHelper& ihelper = helper; + auto subResult = ihelper.subscribe(this, "tts.onSpeechInterrupted", std::move(notification), + onPropertyChangedCallback); + 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); + + // Simulate the platform sending onSpeechInterrupted after unsubscribeAll() — replay the race by calling + // 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 notification = [¬ificationFired](int) { notificationFired = true; }; + + IHelper& ihelper = helper; + auto subResult = + ihelper.subscribe(this, "test.onEvent", std::move(notification), onPropertyChangedCallback); + 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 +}