feat: Add context thread, enqueue all RUM and Logging operations for async processing

src/CMakeLists.txt

@@ -24,6 +24,7 @@ add_library(dd_native ${DD_LIBRARY_TYPE}
     datadog/cpp/version.cpp
     datadog/impl/attribute/cow.cpp
     datadog/impl/attribute/merge.cpp
+    datadog/impl/core/context_thread.cpp
     datadog/impl/core/context.cpp
     datadog/impl/core/core.cpp
     datadog/impl/core/feature_read.cpp

src/datadog/impl/core/context.cpp

@@ -175,6 +175,8 @@ CoreContext::CoreContext(
 )
     : http(std::make_shared<HttpContext>(config)), os(&os_info), device(&device_info) {}
 
+void CoreContext::Reset() { rum.reset(); }
+
 CoreContextProvider::CoreContextProvider(const CoreContext& context)
     : _context(context) {}
 

src/datadog/impl/core/context.hpp

@@ -143,6 +143,13 @@ struct CoreContext {
    * Additional context provided by the RUM feature, if in use.
    */
   std::optional<RumFeatureContext> rum;
+
+  /**
+   * Resets all mutable feature-specific context fields to their default (absent) state.
+   * Called by Core at the start of each run to ensure features never inherit stale
+   * context from a previous run.
+   */
+  void Reset();
 };
 
 /**

src/datadog/impl/core/context_thread.cpp

@@ -0,0 +1,25 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+//
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2025-Present Datadog, Inc.
+
+#include "datadog/impl/core/context_thread.hpp"
+
+namespace datadog::impl {
+
+void ContextThreadMain(
+    const DiagnosticLogger& diagnostic_logger, Queue<std::function<void()>>& queue
+) {
+  diagnostic_logger.Debug("Context thread starting");
+
+  // Process functions until the queue is stopped and drained (Pop() returns
+  // std::nullopt)
+  while (auto thunk = queue.Pop()) {
+    (*thunk)();
+  }
+
+  diagnostic_logger.Debug("Context thread finished");
+}
+
+}  // namespace datadog::impl

src/datadog/impl/core/context_thread.hpp

@@ -0,0 +1,37 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+//
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2025-Present Datadog, Inc.
+
+#pragma once
+
+#include <functional>
+
+#include "datadog/impl/core/context.hpp"
+#include "datadog/impl/core/queue.hpp"
+#include "datadog/impl/diagnostics.hpp"
+
+namespace datadog::impl {
+
+/**
+ * Entry point for the context thread.
+ *
+ * The context thread continually reads functions from the context queue, execute each
+ * function serially as it's consumed. This allows features to perform operations
+ * asynchronously without blocking their callers.
+ *
+ * Functions are enqueued via FeatureScope::ExecuteOnContextThread et al. - FeatureScope
+ * is responsible for wrapping Feature-provided functions in a thunk that will evaluate
+ * the required parameters (CoreContext and EventWriter) from a CoreContextProvider, so
+ * the context thread itself is dead-simple: it just executes those thunks.
+ *
+ * @param diagnostic_logger Interface for logging status/warning messages.
+ * @param queue Non-owning reference to the thread-safe queue that we read from;
+ *  guaranteed to outlive the thread.
+ */
+void ContextThreadMain(
+    const DiagnosticLogger& diagnostic_logger, Queue<std::function<void()>>& queue
+);
+
+}  // namespace datadog::impl

src/datadog/impl/core/core.cpp

@@ -12,6 +12,7 @@
 #include <sstream>
 
 #include "datadog/impl/assert.hpp"
+#include "datadog/impl/core/context_thread.hpp"
 #include "datadog/impl/core/storage_thread.hpp"
 #include "datadog/impl/core/types.hpp"
 #include "datadog/impl/core/version.hpp"
@@ -298,6 +299,11 @@ bool Core::Start() {
     return false;
   }
 
+  // Reset any feature-specific context left over from a previous run, so that features
+  // start clean rather than inheriting stale state from the prior session
+  DATADOG_ASSERT(_context_provider, "_context_provider is null on Start()");
+  _context_provider->Update([](CoreContext& ctx) { ctx.Reset(); });
+
   // Initialize a thread-safe queue that features can write to whenever they produce
   // events that need to be written to disk
   DATADOG_ASSERT(!_storage_queue, "_storage_queue already exists on Start()");
@@ -314,6 +320,19 @@ bool Core::Start() {
       std::ref(_features)
   );
 
+  // Initialize a thread-safe queue for functions that will execute on the context
+  // thread
+  DATADOG_ASSERT(!_context_queue, "_context_queue already exists on Start()");
+  _context_queue = std::make_unique<Queue<std::function<void()>>>();
+
+  // Start the context thread that will execute functions submitted by features
+  DATADOG_ASSERT(!_context_thread, "_context_thread already exists on Start()");
+  _context_thread = std::thread(
+      ContextThreadMain, std::ref(_diagnostic_logger), std::ref(*_context_queue)
+  );
+
+  // Initialize an upload scheduler to manage the timing of upload cycles for each
+  // feature
   DATADOG_ASSERT(!_upload_scheduler, "_upload_scheduler already exists on Start()");
   _upload_scheduler = std::make_unique<UploadScheduler>(*_subsystems.clock);
 
@@ -355,12 +374,13 @@ bool Core::Start() {
   // FeatureScope interface that it can use to interoperate with the core
   for (const auto& feature : _features) {
     const FeatureId id = feature.id;
-    EventGeneratedFunc event_generated_func =
-        [this, id](Block event, Block event_metadata) -> bool {
+    EventWriter event_writer = [this, id](Block event, Block event_metadata) -> bool {
       return EnqueueStorageWrite(id, event, event_metadata);
     };
     feature.impl->OnCoreStarted(
-        FeatureScope(*_context_provider, event_generated_func, _diagnostic_logger)
+        FeatureScope::Create(
+            *_context_provider, event_writer, _diagnostic_logger, *_context_queue
+        )
     );
   }
   return true;
@@ -373,12 +393,12 @@ void Core::Stop() {
   }
   _diagnostic_logger.Debug("Beginning Core shutdown");
 
-  // Notify each registered feature that the core has stopped
-  for (const auto& feature : _features) {
-    feature.impl->OnCoreStopping();
-  }
-
-  // If we were previously started, the storage and upload threads should be running
+  // If we were previously started, all background threads should be running
+  DATADOG_ASSERT(_context_queue, "_context_queue is invalid on Stop");
+  DATADOG_ASSERT(
+      _context_thread && _context_thread->joinable(),
+      "_context_thread is non-joinable on Stop"
+  );
   DATADOG_ASSERT(_storage_queue, "_storage_queue is invalid on Stop");
   DATADOG_ASSERT(
       _storage_thread && _storage_thread->joinable(),
@@ -390,6 +410,27 @@ void Core::Stop() {
       "_upload_thread is non-joinable on Stop"
   );
 
+  // Stop the context queue, then block until the context thread drains the queue and
+  // exits. This ensures all pending feature work completes before we stop the storage
+  // thread or tear down feature state.
+  // TODO(RUM-15042): Other SDKs abandon the context thread and shut down in a
+  // non-blocking fashion, without draining the context queue
+  _context_queue->Stop();
+  if (_context_thread) {
+    _diagnostic_logger.Debug("Joining on context thread");
+    _context_thread->join();
+  }
+  _context_thread.reset();
+
+  // 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
+  _context_queue.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
   _storage_queue->Stop();

src/datadog/impl/core/core.hpp

@@ -386,6 +386,9 @@ class Core {
   std::unique_ptr<StorageQueue> _storage_queue;
   std::optional<std::thread> _storage_thread;
 
+  std::unique_ptr<Queue<std::function<void()>>> _context_queue;
+  std::optional<std::thread> _context_thread;
+
   std::unique_ptr<UploadScheduler> _upload_scheduler;
   std::optional<std::thread> _upload_thread;
 

src/datadog/impl/core/feature.cpp

@@ -20,22 +20,12 @@ void Feature::OnCoreStarted(FeatureScope&& feature_scope) {
 }
 
 void Feature::OnCoreStopping() {
-  // Notify the feature that the core is about to stop, while it's still able to write
-  // events
+  // Notify the feature that the core is stopping
   Stop();
 
   // Drop our FeatureScope; the Core is stopped so we must stop all work
   DATADOG_ASSERT(_scope, "Feature has null _scope in OnCoreStop");
   _scope.reset();
 }
 
-bool Feature::WriteEvent(Block event, Block event_metadata) const {
-  if (_scope) {
-    return _scope->WriteEvent(event, event_metadata);
-  }
-  return false;
-}
-
-bool Feature::IsRunning() const { return _scope.has_value(); }
-
 }  // namespace datadog::impl

src/datadog/impl/core/feature.hpp

@@ -74,8 +74,8 @@ struct FeatureStorageConfig {
  * - If the feature has unique storage requirements, implement `GetStorageConfig()`
  * - If the feature needs initialization/shutdown logic, implement `Start()`/`Stop()`
  * - Define main-thread-callable functions for the required feature-specific operations
- * - In those functions, call `WriteEvent()` for each event the feature generate, using
- *    whatever binary format is appropriate for the feature
+ * - In those functions, call `_scope->ExecuteOnContextThread()` to generate events
+ *    asynchronously on the context thread, using the provided `EventWriter`
  * - Implement `UploadThread_PrepareReport()` to read batches of events (in the same
  *    format) and return a `Report` object describing the resulting HTTP request that
  *    should be made to upload that batch to the appropriate intake endpoint
@@ -110,32 +110,22 @@ class Feature : public std::enable_shared_from_this<Feature> {
  protected:
   /**
    * Called from the main thread when the SDK has finished starting up. This is the
-   * first point at which events may be generated.
+   * first point at which _scope is valid.
+   *
+   * Once Start() is called, the feature may enqueue work on the context thread via
+   * _scope->ExecuteOnContextThread(), generating events in the process, until Stop() is
+   * called.
    */
   virtual void Start() {}
 
   /**
-   * Called from the main thread when the SDK is about to shut down. This is the last
-   * point at which events may be generated.
-   */
-  virtual void Stop() {}
-
-  /**
-   * Callable from the main thread in order to produce a new event. Copies the given
-   * event payload(s) into the storage queue so that the storage thread can write them
-   * to persistent storage. Once the batch containing this event is ready for upload,
-   * the upload thread will pass it to UploadThread_PrepareReport().
+   * Called from the main thread when the SDK is shutting down. This is the last point
+   * at which _scope is valid.
    *
-   * @returns whether the event was successfully enqueued for storage. If called before
-   *  Start() or after Stop(), always returns false.
+   * Once Stop() is called, any work enqueued via _scope->ExecuteOnContextThread() will
+   * be ignored, and the feature may no longer generate events.
    */
-  bool WriteEvent(Block event, Block event_metadata = {}) const;
-
-  /**
-   * @returns whether the feature has received a call to Start() and has not yet
-   *  received a call to Stop().
-   */
-  bool IsRunning() const;
+  virtual void Stop() {}
 
  public:
   /**

src/datadog/impl/core/feature_scope.cpp

@@ -12,28 +12,89 @@ namespace datadog::impl {
 
 FeatureScope::FeatureScope(
     CoreContextProvider& context_provider,
-    const EventGeneratedFunc& event_generated_func,
-    const DiagnosticLogger& in_diagnostic_logger
+    const EventWriter& event_writer,
+    const DiagnosticLogger& in_diagnostic_logger,
+    FeatureScope::ExecutionMode mode,
+    Queue<std::function<void()>>* context_queue
 )
     : _context_provider(&context_provider),
-      _event_generated_func(event_generated_func),
-      diagnostic_logger(in_diagnostic_logger) {}
+      _event_generated_func(event_writer),
+      _mode(mode),
+      _context_queue(context_queue),
+      diagnostic_logger(in_diagnostic_logger) {
+  // Invariant: context_queue must be non-null when mode is OnContextThread
+  DATADOG_ASSERT(
+      mode != FeatureScope::ExecutionMode::OnContextThread || context_queue != nullptr,
+      "context_queue must be non-null when ExecutionMode is OnContextThread"
+  );
+}
 
-CoreContext FeatureScope::GetContext() const {
-  DATADOG_ASSERT(_context_provider, "FeatureScope has no _context_provider");
-  return _context_provider->Get();
+FeatureScope FeatureScope::Create(
+    CoreContextProvider& context_provider,
+    const EventWriter& event_writer,
+    const DiagnosticLogger& diagnostic_logger,
+    Queue<std::function<void()>>& context_queue
+) {
+  return FeatureScope(
+      context_provider,
+      event_writer,
+      diagnostic_logger,
+      FeatureScope::ExecutionMode::OnContextThread,
+      &context_queue
+  );
+}
+
+FeatureScope FeatureScope::CreateForTesting(
+    CoreContextProvider& context_provider,
+    const EventWriter& event_writer,
+    const DiagnosticLogger& diagnostic_logger
+) {
+  return FeatureScope(
+      context_provider,
+      event_writer,
+      diagnostic_logger,
+      FeatureScope::ExecutionMode::Synchronous,
+      nullptr
+  );
 }
 
 void FeatureScope::UpdateContext(const std::function<void(CoreContext&)>& callback) {
   DATADOG_ASSERT(_context_provider, "FeatureScope has no _context_provider");
-  _context_provider->Update(callback);
+
+  if (_mode == FeatureScope::ExecutionMode::Synchronous) {
+    // Testing mode: execute synchronously on calling thread
+    _context_provider->Update(callback);
+    return;
+  }
+
+  // Production mode: queue for async execution on context thread
+  DATADOG_ASSERT(
+      _context_queue != nullptr, "context_queue is null in OnContextThread mode"
+  );
+  _context_queue->Push([this, callback]() {
+    DATADOG_ASSERT(_context_provider, "FeatureScope has no _context_provider");
+    _context_provider->Update(callback);
+  });
 }
 
-bool FeatureScope::WriteEvent(Block event, Block event_metadata) const {
-  if (_event_generated_func) {
-    return _event_generated_func(event, event_metadata);
+void FeatureScope::ExecuteOnContextThread(const ContextThreadFunc& func) {
+  if (_mode == FeatureScope::ExecutionMode::Synchronous) {
+    // Testing mode: execute synchronously on calling thread
+    DATADOG_ASSERT(_context_provider, "FeatureScope has no _context_provider");
+    const CoreContext context = _context_provider->Get();
+    func(context, _event_generated_func);
+    return;
   }
-  return false;
+
+  // Production mode: queue for async execution on context thread
+  DATADOG_ASSERT(
+      _context_queue != nullptr, "context_queue is null in OnContextThread mode"
+  );
+  _context_queue->Push([this, func]() {
+    DATADOG_ASSERT(_context_provider, "FeatureScope has no _context_provider");
+    const CoreContext context = _context_provider->Get();
+    func(context, _event_generated_func);
+  });
 }
 
 }  // namespace datadog::impl