chore: auto-sync develop → main (direct push failure)

.github/workflows/release.yml

@@ -17,7 +17,7 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          token: ${{ secrets.SEMANTIC_RELEASE_TOKEN }}
+          token: ${{ secrets.AUTOMATION_TOKEN }}
 
       - name: Setup Node.js
         uses: actions/setup-node@v4
@@ -30,4 +30,4 @@ jobs:
       - name: Run semantic-release
         run: npm --prefix .github/release exec -- semantic-release --extends ./.github/release/releaserc.json
         env:
-          GITHUB_TOKEN: ${{ secrets.SEMANTIC_RELEASE_TOKEN }}
+          GITHUB_TOKEN: ${{ secrets.AUTOMATION_TOKEN }}

.github/workflows/sync-develop-to-main.yml

@@ -37,7 +37,7 @@ jobs:
       - name: Checkout full history
         uses: actions/checkout@v4
         with:
-          token: ${{ secrets.SEMANTIC_RELEASE_TOKEN }}
+          token: ${{ secrets.AUTOMATION_TOKEN }}
           fetch-depth: 0
 
       - name: Configure git identity
@@ -86,7 +86,7 @@ jobs:
       - name: Open PR on conflict
         if: steps.check.outputs.already_synced == 'false' && (steps.merge.outputs.conflict == 'true' || steps.push_merged.outputs.push_failed == 'true')
         env:
-          GH_TOKEN: ${{ secrets.SEMANTIC_RELEASE_TOKEN }}
+          GH_TOKEN: ${{ secrets.AUTOMATION_TOKEN }}
         run: |
           PR_BRANCH="auto-sync/${{ env.SOURCE_BRANCH }}-to-${{ env.TARGET_BRANCH }}"
           REASON="merge conflict"

include/firebolt/helpers.h

@@ -40,10 +40,10 @@ template <typename JsonType, typename... Args>
 void onPropertyChangedCallback(void* subscriptionDataPtr, const nlohmann::json& jsonResponse)
 {
     SubscriptionData* subscriptionData = reinterpret_cast<SubscriptionData*>(subscriptionDataPtr);
-    auto notifier = std::any_cast<std::function<void(Args...)>>(subscriptionData->notification);
-    JsonType jsonType;
     try
     {
+        auto notifier = std::any_cast<std::function<void(Args...)>>(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());
     }
 }

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.