From 48b0638df69c7e3d86b4750da2aa1d6e292e9dba Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Wed, 11 Mar 2026 15:40:58 -0400 Subject: [PATCH 1/9] feat: Add `MessageBus` to core, add `MakeMessageHandler` to `Feature` refs: RUM-12206 --- src/datadog/impl/core/context.cpp | 21 +++- src/datadog/impl/core/context.hpp | 16 +++ src/datadog/impl/core/core.cpp | 39 ++++++- src/datadog/impl/core/core.hpp | 6 + src/datadog/impl/core/feature.hpp | 17 +++ tests/impl/core/message_bus_test.cpp | 168 +++++++++++++++++++++++++++ 6 files changed, 263 insertions(+), 4 deletions(-) diff --git a/src/datadog/impl/core/context.cpp b/src/datadog/impl/core/context.cpp index cd16055c..5d082d2e 100644 --- a/src/datadog/impl/core/context.cpp +++ b/src/datadog/impl/core/context.cpp @@ -12,6 +12,7 @@ #include "datadog/uuid.hpp" #include "datadog/impl/assert.hpp" +#include "datadog/impl/core/message_bus.hpp" #include "datadog/impl/core/version.hpp" namespace datadog::impl { @@ -187,11 +188,25 @@ CoreContext CoreContextProvider::Get() const { } void CoreContextProvider::Update(const std::function& callback) { - // Acquire an exclusive write lock, then allow the callback to mutate the context - std::unique_lock lock(_mutex); - callback(_context); + // Acquire an exclusive write lock, mutate the context, and capture a snapshot. The + // lock is released before Send() to minimize contention on the message bus. + std::optional snapshot; + { + std::unique_lock lock(_mutex); + callback(_context); + snapshot = _context; + } + + // CoreContext has been mutated (and we have a valid snapshot): notify all registered + // message-handlers of the change, so that Features can perform work (on the messaging + // thread) in response to the updated context + if (_message_bus) { + _message_bus->Send(ContextChangedMessage{std::move(*snapshot)}); + } } +void CoreContextProvider::SetMessageBus(MessageBus* bus) { _message_bus = bus; } + const HttpContext& CoreContextProvider::GetHttpContext() const { std::shared_lock lock(_mutex); return *_context.http; diff --git a/src/datadog/impl/core/context.hpp b/src/datadog/impl/core/context.hpp index d66c6407..1d2daac6 100644 --- a/src/datadog/impl/core/context.hpp +++ b/src/datadog/impl/core/context.hpp @@ -24,6 +24,9 @@ struct DeviceInfo; namespace datadog::impl { +// Forward declaration — full type used only in context.cpp +class MessageBus; + /** * SDK configuration details that influence how HTTP requests are built. Immutable for * the lifetime of the Core. @@ -159,6 +162,7 @@ class CoreContextProvider { private: CoreContext _context; mutable std::shared_mutex _mutex; + MessageBus* _message_bus{nullptr}; public: explicit CoreContextProvider(const CoreContext& context); @@ -171,6 +175,10 @@ class CoreContextProvider { /** * Mutates the current CoreContext value, by invoking the provided callback after * obtaining exclusive write access. + * + * After mutation, dispatches a `ContextChangedMessage` on the `MessageBus`, carrying + * an immutable snapshot of the updated context to any Features that have registered + * an interest in receiving notifications of context changes. */ void Update(const std::function& callback); @@ -178,6 +186,14 @@ class CoreContextProvider { * Synchronously returns an immutable reference to the HttpContext value. */ const HttpContext& GetHttpContext() const; + + /** + * Registers the `MessageBus` that `Update()` will dispatch `ContextChangedMessage` + * values to. Pass `nullptr` to detach the bus (e.g. during shutdown). Called by + * `Core::Start()` before the context thread launches; no synchronization is needed + * because the context thread is not yet running at that point. + */ + void SetMessageBus(MessageBus* bus); }; } // namespace datadog::impl diff --git a/src/datadog/impl/core/core.cpp b/src/datadog/impl/core/core.cpp index ac976702..ec1e3d06 100644 --- a/src/datadog/impl/core/core.cpp +++ b/src/datadog/impl/core/core.cpp @@ -266,6 +266,14 @@ bool Core::RegisterFeature(const std::shared_ptr& impl) { std::move(*event_read_directory), std::move(upload_state) ); + + // Collect the feature's message handler, if it provides one. weak_from_this() is + // safe to call here because the feature has been stored in _features and therefore + // already has shared ownership. + if (auto h = impl->MakeMessageHandler()) { + _pending_handlers.push_back(std::move(*h)); + } + _diagnostic_logger.Debug( "Feature registered", {{"feature", name}, {"feature_id", static_cast(id)}} @@ -320,6 +328,18 @@ bool Core::Start() { std::ref(_features) ); + // Construct the message bus from any handlers registered by features, wire it into + // the context provider so Update() will dispatch ContextChangedMessage values, and + // start the messaging thread. This is done before the context thread launches so + // that SetMessageBus() requires no synchronization. + DATADOG_ASSERT(!_message_bus, "_message_bus already exists on Start()"); + _message_bus = std::make_unique(std::move(_pending_handlers)); + _context_provider->SetMessageBus(_message_bus.get()); + DATADOG_ASSERT(!_message_bus_thread, "_message_bus_thread already exists on Start()"); + _message_bus_thread = std::thread( + MessagingThreadMain, std::ref(_diagnostic_logger), std::ref(*_message_bus) + ); + // Initialize a thread-safe queue for functions that will execute on the context // thread DATADOG_ASSERT(!_context_queue, "_context_queue already exists on Start()"); @@ -422,14 +442,31 @@ void Core::Stop() { } _context_thread.reset(); + // Stop the messaging thread after the context thread exits: the context thread is + // the primary producer of messages (via Update()), so draining it first prevents new + // messages from arriving after we signal the messaging thread to stop. After joining, + // detach the bus from the context provider so any stray Update() calls (which should + // not happen at this point) do not reference the destroyed bus. + DATADOG_ASSERT(_message_bus, "_message_bus is invalid on Stop"); + DATADOG_ASSERT( + _message_bus_thread && _message_bus_thread->joinable(), + "_message_bus_thread is non-joinable on Stop" + ); + _message_bus->Stop(); + _diagnostic_logger.Debug("Joining on messaging thread"); + _message_bus_thread->join(); + _message_bus_thread.reset(); + _context_provider->SetMessageBus(nullptr); + // Notify each registered feature that the core has stopped, now that no more // context-thread functions enqueued by those features may be running for (const auto& feature : _features) { feature.impl->OnCoreStopping(); } - // Destroy the context queue, now that no more features exist + // Destroy the context queue and message bus, now that no more features exist _context_queue.reset(); + _message_bus.reset(); // Stop all queue processing, then block until the consumer thread drains the queue // and exits, at which point it's safe to release the queue diff --git a/src/datadog/impl/core/core.hpp b/src/datadog/impl/core/core.hpp index 94e33ca0..4882980a 100644 --- a/src/datadog/impl/core/core.hpp +++ b/src/datadog/impl/core/core.hpp @@ -20,6 +20,8 @@ #include "datadog/impl/core/block.hpp" #include "datadog/impl/core/context.hpp" #include "datadog/impl/core/feature.hpp" +#include "datadog/impl/core/message_bus.hpp" +#include "datadog/impl/core/messaging_thread.hpp" #include "datadog/impl/core/queue.hpp" #include "datadog/impl/core/storage_queue.hpp" #include "datadog/impl/core/storage_write.hpp" @@ -381,6 +383,7 @@ class Core { // Initialized before Start in response to user-initiated feature registration std::vector _features; // May not be modified while running + std::vector> _pending_handlers; // Initialized on Start, cleaned up on Stop std::unique_ptr _storage_queue; @@ -389,6 +392,9 @@ class Core { std::unique_ptr>> _context_queue; std::optional _context_thread; + std::unique_ptr _message_bus; + std::optional _message_bus_thread; + std::unique_ptr _upload_scheduler; std::optional _upload_thread; diff --git a/src/datadog/impl/core/feature.hpp b/src/datadog/impl/core/feature.hpp index 6ee1d4e3..5c0d320d 100644 --- a/src/datadog/impl/core/feature.hpp +++ b/src/datadog/impl/core/feature.hpp @@ -18,6 +18,7 @@ #include "datadog/impl/core/block.hpp" #include "datadog/impl/core/context.hpp" #include "datadog/impl/core/feature_id.hpp" +#include "datadog/impl/core/feature_message.hpp" #include "datadog/impl/core/feature_scope.hpp" #include "datadog/impl/core/tlv.hpp" #include "datadog/impl/platform/filesystem.hpp" @@ -136,6 +137,22 @@ class Feature : public std::enable_shared_from_this { const HttpContext& context, class BatchReader& reader ) = 0; + /** + * Called by `Core::RegisterFeature()` after the feature has been stored in the + * feature list, at which point `weak_from_this()` is safe to call. If the feature + * wants to react to messages dispatched by the `MessageBus`, it should override this + * method and return a handler lambda: typically one that captures `weak_from_this()` + * so that message delivery is silently skipped if the feature has already been + * destroyed. + * + * The default implementation returns `std::nullopt`, meaning the feature opts out of + * message handling entirely. + */ + virtual std::optional> + MakeMessageHandler() { + return std::nullopt; + } + protected: std::optional _scope; }; diff --git a/tests/impl/core/message_bus_test.cpp b/tests/impl/core/message_bus_test.cpp index 26b6248b..39b730cf 100644 --- a/tests/impl/core/message_bus_test.cpp +++ b/tests/impl/core/message_bus_test.cpp @@ -11,6 +11,8 @@ #include #include +#include "datadog/impl/core/feature.hpp" +#include "datadog/impl/core/feature_id.hpp" #include "datadog/impl/core/feature_message.hpp" #include "datadog/impl/core/messaging_thread.hpp" @@ -45,6 +47,172 @@ static uint64_t ReadSentinel(const ContextChangedMessage& msg) { return value; } +/** + * Minimal Feature subclass used to test MakeMessageHandler() wiring. Writes delivered + * ContextChangedMessage values into an external vector whose lifetime exceeds the + * feature's, which allows test assertions even when the feature is destroyed early. + */ +struct MessageBusTestFeature final : public Feature { + std::vector& received; + + explicit MessageBusTestFeature(std::vector& out) + : received(out) {} + + FeatureId GetId() const override { return CreateFeatureId("MBUS"); } + std::string_view GetName() const override { return "message_bus_test"; } + std::optional UploadThread_PrepareReport( + const HttpContext&, class BatchReader& + ) override { + return std::nullopt; + } + + std::optional> + MakeMessageHandler() override { + // Capture both the weak_ptr (for lifetime gating) and a direct reference to the + // output vector (which outlives the Feature in tests). Since `received` is a + // reference member, `&out` in the lambda binds to the external vector, not to + // this object, so the reference remains valid after the Feature is destroyed. + auto weak = weak_from_this(); + std::vector& out = received; + return [weak, &out](const FeatureMessage& msg) { + if (!weak.lock()) { + return; + } + out.push_back(std::get(msg)); + }; + } +}; + +TEST_CASE("MessageBus — integration", "[unit][core]") { + DiagnosticMessageBuffer diagnostics; + DiagnosticLogger logger = diagnostics.CreateTestLogger(); + + SECTION( + "M deliver ContextChangedMessage W MakeMessageHandler() wired through " + "CoreContextProvider::Update()" + ) { + // Given a feature registered via MakeMessageHandler() + std::vector received; + auto feature = std::make_shared(received); + auto handler = feature->MakeMessageHandler(); + REQUIRE(handler.has_value()); + + std::vector> handlers; + handlers.push_back(std::move(*handler)); + + MessageBus bus(std::move(handlers)); + std::thread t(MessagingThreadMain, std::ref(logger), std::ref(bus)); + + CoreContextProvider context_provider(MOCK_CONTEXT); + context_provider.SetMessageBus(&bus); + + // When Update() is called + context_provider.Update([](CoreContext& ctx) { ctx.rum.emplace(); }); + + // Then the handler receives the message + bus.Stop(); + t.join(); + context_provider.SetMessageBus(nullptr); + + REQUIRE(received.size() == 1); + } + + SECTION("M deliver correct context snapshot W context mutated in Update()") { + // Given a bus with a plain lambda handler + std::vector received; + std::vector> handlers; + handlers.push_back([&](const FeatureMessage& msg) { + received.push_back(std::get(msg)); + }); + + MessageBus bus(std::move(handlers)); + std::thread t(MessagingThreadMain, std::ref(logger), std::ref(bus)); + + CoreContextProvider context_provider(MOCK_CONTEXT); + context_provider.SetMessageBus(&bus); + + // When Update() writes a known sentinel value into the RUM session_id + constexpr uint64_t sentinel = 0xc0ffee; + context_provider.Update([](CoreContext& ctx) { + ctx.rum.emplace(); + constexpr uint64_t s = 0xc0ffee; + std::memcpy(ctx.rum->session_id.bytes.data(), &s, sizeof(s)); + }); + + bus.Stop(); + t.join(); + context_provider.SetMessageBus(nullptr); + + // Then the delivered snapshot carries that exact value + REQUIRE(received.size() == 1); + REQUIRE(received[0].context.rum.has_value()); + uint64_t got = 0; + std::memcpy(&got, received[0].context.rum->session_id.bytes.data(), sizeof(got)); + REQUIRE(got == sentinel); + } + + SECTION( + "M skip delivery gracefully W feature is destroyed before message is handled" + ) { + // Given a feature whose handler captures weak_from_this(), and a second handler + // that counts deliveries to confirm the bus continues to function + std::vector received; + auto feature = std::make_shared(received); + auto handler = feature->MakeMessageHandler(); + REQUIRE(handler.has_value()); + + std::atomic other_count{0}; + std::vector> handlers; + handlers.push_back(std::move(*handler)); + handlers.push_back([&](const FeatureMessage&) { other_count.fetch_add(1); }); + + // Enqueue a message while the feature is still alive, then destroy it before the + // messaging thread has a chance to drain the queue. Running the thread function + // synchronously (after Stop()) guarantees the feature is gone when the handler + // runs. + MessageBus bus(std::move(handlers)); + REQUIRE(bus.Send(MakeMessage(0x1))); + + feature.reset(); // weak_from_this() in handler now expires + + bus.Stop(); + MessagingThreadMain(logger, bus); // drain synchronously; no background thread + + // The TestFeature handler silently dropped the message; the other handler ran + REQUIRE(received.empty()); + REQUIRE(other_count.load() == 1); + } + + SECTION( + "M deliver all messages enqueued before Stop() W shutdown via " + "CoreContextProvider::Update()" + ) { + // Given a bus wired to a context provider with a counting handler + std::atomic delivered{0}; + std::vector> handlers; + handlers.push_back([&](const FeatureMessage&) { delivered.fetch_add(1); }); + + MessageBus bus(std::move(handlers)); + std::thread t(MessagingThreadMain, std::ref(logger), std::ref(bus)); + + CoreContextProvider context_provider(MOCK_CONTEXT); + context_provider.SetMessageBus(&bus); + + // When multiple Update() calls are made before the bus is stopped + const int num_updates = 20; + for (int i = 0; i < num_updates; i++) { + context_provider.Update([](CoreContext& ctx) { ctx.rum.emplace(); }); + } + + bus.Stop(); + t.join(); + context_provider.SetMessageBus(nullptr); + + // Then all messages were delivered before the join returned + REQUIRE(delivered.load() == num_updates); + } +} + TEST_CASE("MessageBus", "[unit][core]") { DiagnosticMessageBuffer diagnostics; DiagnosticLogger logger = diagnostics.CreateTestLogger(); From 73c793dee3c07782e23d409a45b8a83291bd8091 Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Wed, 11 Mar 2026 15:47:22 -0400 Subject: [PATCH 2/9] chore: Fix redeclared sentinel in test --- tests/impl/core/message_bus_test.cpp | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/tests/impl/core/message_bus_test.cpp b/tests/impl/core/message_bus_test.cpp index 39b730cf..a4615457 100644 --- a/tests/impl/core/message_bus_test.cpp +++ b/tests/impl/core/message_bus_test.cpp @@ -133,10 +133,9 @@ TEST_CASE("MessageBus — integration", "[unit][core]") { // When Update() writes a known sentinel value into the RUM session_id constexpr uint64_t sentinel = 0xc0ffee; - context_provider.Update([](CoreContext& ctx) { + context_provider.Update([=](CoreContext& ctx) { ctx.rum.emplace(); - constexpr uint64_t s = 0xc0ffee; - std::memcpy(ctx.rum->session_id.bytes.data(), &s, sizeof(s)); + std::memcpy(ctx.rum->session_id.bytes.data(), &sentinel, sizeof(sentinel)); }); bus.Stop(); From d461f48959937eed3e1d97a09eebc797edb174ac Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Wed, 11 Mar 2026 15:49:17 -0400 Subject: [PATCH 3/9] chore: Move messaging_thread.hpp include to core.cpp --- src/datadog/impl/core/core.cpp | 1 + src/datadog/impl/core/core.hpp | 1 - 2 files changed, 1 insertion(+), 1 deletion(-) diff --git a/src/datadog/impl/core/core.cpp b/src/datadog/impl/core/core.cpp index ec1e3d06..e5f90ae2 100644 --- a/src/datadog/impl/core/core.cpp +++ b/src/datadog/impl/core/core.cpp @@ -13,6 +13,7 @@ #include "datadog/impl/assert.hpp" #include "datadog/impl/core/context_thread.hpp" +#include "datadog/impl/core/messaging_thread.hpp" #include "datadog/impl/core/storage_thread.hpp" #include "datadog/impl/core/types.hpp" #include "datadog/impl/core/version.hpp" diff --git a/src/datadog/impl/core/core.hpp b/src/datadog/impl/core/core.hpp index 4882980a..d6064be4 100644 --- a/src/datadog/impl/core/core.hpp +++ b/src/datadog/impl/core/core.hpp @@ -21,7 +21,6 @@ #include "datadog/impl/core/context.hpp" #include "datadog/impl/core/feature.hpp" #include "datadog/impl/core/message_bus.hpp" -#include "datadog/impl/core/messaging_thread.hpp" #include "datadog/impl/core/queue.hpp" #include "datadog/impl/core/storage_queue.hpp" #include "datadog/impl/core/storage_write.hpp" From 90b4c8c3c163b87683379acc692fffeafd0dcc89 Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Wed, 11 Mar 2026 15:51:32 -0400 Subject: [PATCH 4/9] chore: Use IIFE for snapshot rather than mutating std::optional --- src/datadog/impl/core/context.cpp | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/src/datadog/impl/core/context.cpp b/src/datadog/impl/core/context.cpp index 5d082d2e..575fb939 100644 --- a/src/datadog/impl/core/context.cpp +++ b/src/datadog/impl/core/context.cpp @@ -190,18 +190,17 @@ CoreContext CoreContextProvider::Get() const { void CoreContextProvider::Update(const std::function& callback) { // Acquire an exclusive write lock, mutate the context, and capture a snapshot. The // lock is released before Send() to minimize contention on the message bus. - std::optional snapshot; - { + CoreContext snapshot = [&] { std::unique_lock lock(_mutex); callback(_context); - snapshot = _context; - } + return _context; + }(); // CoreContext has been mutated (and we have a valid snapshot): notify all registered // message-handlers of the change, so that Features can perform work (on the messaging // thread) in response to the updated context if (_message_bus) { - _message_bus->Send(ContextChangedMessage{std::move(*snapshot)}); + _message_bus->Send(ContextChangedMessage{std::move(snapshot)}); } } From 39a53d1b5747f303b3f8debd886edbca5dbfe7be Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Wed, 11 Mar 2026 16:02:50 -0400 Subject: [PATCH 5/9] chore: Add a failing test to catch lost message handler on Core restart --- tests/impl/core/core_test.cpp | 58 +++++++++++++++++++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/tests/impl/core/core_test.cpp b/tests/impl/core/core_test.cpp index 110529f5..1634866b 100644 --- a/tests/impl/core/core_test.cpp +++ b/tests/impl/core/core_test.cpp @@ -6,8 +6,11 @@ #include "datadog/impl/core/core.hpp" +#include #include +#include "datadog/impl/core/feature_message.hpp" + #include "mock/clock.hpp" #include "mock/feature.hpp" #include "mock/filesystem.hpp" @@ -288,3 +291,58 @@ TEST_CASE("Core Lifecycle", "[unit]") { REQUIRE(feature->num_stop_calls == 1); } } + +/** + * Feature that counts received ContextChangedMessages via MakeMessageHandler(), and + * triggers a context update (causing one such message to be dispatched) each time the + * core starts. + */ +class MessageHandlerFeature : public MockFeature { + public: + std::atomic messages_received{0}; + + MessageHandlerFeature() + : MockFeature(CreateFeatureId("MHFT"), "message_handler_test") {} + + std::optional> + MakeMessageHandler() override { + auto weak = weak_from_this(); + return [weak](const FeatureMessage&) { + if (auto self = std::static_pointer_cast(weak.lock())) { + self->messages_received.fetch_add(1); + } + }; + } + + void Start() override { + MockFeature::Start(); + // Trigger a context update so that a ContextChangedMessage is dispatched to + // all registered handlers on each start + _scope->UpdateContext([](CoreContext& ctx) { ctx.rum.emplace(); }); + } +}; + +TEST_CASE("Core Messaging", "[unit]") { + SECTION( + "M deliver messages to handler after restart W core is stopped and " + "started again" + ) { + impl::Core core = _make_core(); + REQUIRE(core.Init()); + + auto feature = std::make_shared(); + REQUIRE(core.RegisterFeature(feature)); + + // First run: start triggers one context update, stop drains both the context + // thread and the messaging thread before returning + REQUIRE(core.Start()); + core.Stop(); + REQUIRE(feature->messages_received.load() == 1); + + // Second run: the handler must still be registered after the restart; the + // same context update should deliver a second message + REQUIRE(core.Start()); + core.Stop(); + REQUIRE(feature->messages_received.load() == 2); + } +} From 22d0dc5db9aafe0fdb2e6ad2a0a0664f7e077959 Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Wed, 11 Mar 2026 16:10:33 -0400 Subject: [PATCH 6/9] fix: Move per-Feature message handler creation to Core::Start() --- src/datadog/impl/core/core.cpp | 37 +++++++++++++++++++++------------- src/datadog/impl/core/core.hpp | 1 - 2 files changed, 23 insertions(+), 15 deletions(-) diff --git a/src/datadog/impl/core/core.cpp b/src/datadog/impl/core/core.cpp index e5f90ae2..88c94c96 100644 --- a/src/datadog/impl/core/core.cpp +++ b/src/datadog/impl/core/core.cpp @@ -268,13 +268,6 @@ bool Core::RegisterFeature(const std::shared_ptr& impl) { std::move(upload_state) ); - // Collect the feature's message handler, if it provides one. weak_from_this() is - // safe to call here because the feature has been stored in _features and therefore - // already has shared ownership. - if (auto h = impl->MakeMessageHandler()) { - _pending_handlers.push_back(std::move(*h)); - } - _diagnostic_logger.Debug( "Feature registered", {{"feature", name}, {"feature_id", static_cast(id)}} @@ -329,20 +322,36 @@ bool Core::Start() { std::ref(_features) ); - // Construct the message bus from any handlers registered by features, wire it into - // the context provider so Update() will dispatch ContextChangedMessage values, and - // start the messaging thread. This is done before the context thread launches so - // that SetMessageBus() requires no synchronization. + // Iterate through all registered features and call MakeMessageHandler(), allowing + // features to register their intent to be notified when messages are sent on the + // message bus + std::vector> handlers; + handlers.reserve(_features.size()); + for (const auto& f : _features) { + if (auto h = f.impl->MakeMessageHandler()) { + handlers.push_back(std::move(*h)); + } + } + + // Create the message bus, which will queue all core-to-feature and feature-to-feature + // messages that need to be handled by interested features DATADOG_ASSERT(!_message_bus, "_message_bus already exists on Start()"); - _message_bus = std::make_unique(std::move(_pending_handlers)); + _message_bus = std::make_unique(std::move(handlers)); + + // Install the MessageBus into the CoreContextProvider so that it can send + // ContextChangedMessage in response to updates: context thread doesn't exist yet, so + // no synchronization is required here _context_provider->SetMessageBus(_message_bus.get()); + + // Start the messaging thread, which will invoke relevant message handler functions + // for each message sent on the message bus DATADOG_ASSERT(!_message_bus_thread, "_message_bus_thread already exists on Start()"); _message_bus_thread = std::thread( MessagingThreadMain, std::ref(_diagnostic_logger), std::ref(*_message_bus) ); - // Initialize a thread-safe queue for functions that will execute on the context - // thread + // Initialize a thread-safe queue for feature-submitted functions that will execute on + // the context thread DATADOG_ASSERT(!_context_queue, "_context_queue already exists on Start()"); _context_queue = std::make_unique>>(); diff --git a/src/datadog/impl/core/core.hpp b/src/datadog/impl/core/core.hpp index d6064be4..80a06eab 100644 --- a/src/datadog/impl/core/core.hpp +++ b/src/datadog/impl/core/core.hpp @@ -382,7 +382,6 @@ class Core { // Initialized before Start in response to user-initiated feature registration std::vector _features; // May not be modified while running - std::vector> _pending_handlers; // Initialized on Start, cleaned up on Stop std::unique_ptr _storage_queue; From ea485aa7d25d30889bba370f0960baaffc7cb44d Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Thu, 12 Mar 2026 07:56:54 -0400 Subject: [PATCH 7/9] fix: Start message thread after Start() for consistency --- src/datadog/impl/core/core.cpp | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/src/datadog/impl/core/core.cpp b/src/datadog/impl/core/core.cpp index 88c94c96..13957fe9 100644 --- a/src/datadog/impl/core/core.cpp +++ b/src/datadog/impl/core/core.cpp @@ -343,13 +343,6 @@ bool Core::Start() { // no synchronization is required here _context_provider->SetMessageBus(_message_bus.get()); - // Start the messaging thread, which will invoke relevant message handler functions - // for each message sent on the message bus - DATADOG_ASSERT(!_message_bus_thread, "_message_bus_thread already exists on Start()"); - _message_bus_thread = std::thread( - MessagingThreadMain, std::ref(_diagnostic_logger), std::ref(*_message_bus) - ); - // Initialize a thread-safe queue for feature-submitted functions that will execute on // the context thread DATADOG_ASSERT(!_context_queue, "_context_queue already exists on Start()"); @@ -413,6 +406,18 @@ bool Core::Start() { ) ); } + + // Start the messaging thread only after all features have completed OnCoreStarted(). + // This ensures that any ContextChangedMessages dispatched during startup (e.g. from + // a feature calling UpdateContext in its Start()) are not delivered to a feature's + // handler before that feature's own OnCoreStarted() has run. Messages enqueued + // during the loop above are buffered in the queue and will be drained once this + // thread starts. + DATADOG_ASSERT(!_message_bus_thread, "_message_bus_thread already exists on Start()"); + _message_bus_thread = std::thread( + MessagingThreadMain, std::ref(_diagnostic_logger), std::ref(*_message_bus) + ); + return true; } From 2317e653fb513092b1bba407e04353ce3f9d70cf Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Fri, 13 Mar 2026 09:01:10 -0400 Subject: [PATCH 8/9] chore: Who puts em dashes in source code, seriously --- src/datadog/impl/core/context.hpp | 1 - 1 file changed, 1 deletion(-) diff --git a/src/datadog/impl/core/context.hpp b/src/datadog/impl/core/context.hpp index 1d2daac6..8ac38ac1 100644 --- a/src/datadog/impl/core/context.hpp +++ b/src/datadog/impl/core/context.hpp @@ -24,7 +24,6 @@ struct DeviceInfo; namespace datadog::impl { -// Forward declaration — full type used only in context.cpp class MessageBus; /** From 1a908eddf9f3085a8c1475b9dea199b5c290ac74 Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Fri, 13 Mar 2026 09:04:21 -0400 Subject: [PATCH 9/9] chore: Fix outdated comment on MakeMessageHandler --- src/datadog/impl/core/feature.hpp | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/datadog/impl/core/feature.hpp b/src/datadog/impl/core/feature.hpp index 5c0d320d..58385529 100644 --- a/src/datadog/impl/core/feature.hpp +++ b/src/datadog/impl/core/feature.hpp @@ -138,13 +138,14 @@ class Feature : public std::enable_shared_from_this { ) = 0; /** - * Called by `Core::RegisterFeature()` after the feature has been stored in the - * feature list, at which point `weak_from_this()` is safe to call. If the feature - * wants to react to messages dispatched by the `MessageBus`, it should override this - * method and return a handler lambda: typically one that captures `weak_from_this()` + * Called by `Core::Start()` each time the SDK starts up. If the feature wants to + * react to messages dispatched on the `MessageBus`, it should override this method + * and return a handler function: typically a lambda that captures `weak_from_this()` * so that message delivery is silently skipped if the feature has already been * destroyed. * + * Implementations must be idempotent, as this function called on every `Start()`. + * * The default implementation returns `std::nullopt`, meaning the feature opts out of * message handling entirely. */