fix: Add repl flush-work command

examples/repl/commands.hpp

@@ -12,6 +12,7 @@ struct State;
 
 // util
 CommandResult HandleSource(State& state, const CommandInput& args);
+CommandResult HandleFlushWork(State& state, const CommandInput& args);
 CommandResult HandleSleep(State& state, const CommandInput& args);
 CommandResult HandleUrl(State& state, const CommandInput& args);
 CommandResult HandleNop(State& state, const CommandInput& args);

examples/repl/commands_util.cpp

@@ -9,6 +9,8 @@
 #include <iostream>
 #include <thread>
 
+#include "datadog.hpp"
+
 #include "repl/buffer.hpp"
 #include "repl/commands.hpp"
 #include "repl/state.hpp"
@@ -47,6 +49,14 @@ CommandResult HandleSource(State& state, const CommandInput& args) {
   return CommandResult::OK("std::ifstream::open()");
 }
 
+CommandResult HandleFlushWork(State& state, const CommandInput&) {
+  if (!state.core) {
+    return CommandResult::Error("Core does not exist!");
+  }
+  state.core->Internal_FlushWork();
+  return CommandResult::OK("Core::Internal_FlushWork()");
+}
+
 CommandResult HandleSleep(State&, const CommandInput& args) {
   auto pos = args.Positional();
   int interval_ms = 0;

examples/repl/main.cpp

@@ -30,6 +30,9 @@ CommandResult Handle(State& state, const CommandInput& input) {
   if (input.Peek() == "source") {
     return HandleSource(state, input.Shift());
   }
+  if (input.Peek() == "flush-work") {
+    return HandleFlushWork(state, input.Shift());
+  }
   if (input.Peek() == "sleep") {
     return HandleSleep(state, input.Shift());
   }

include-cpp/datadog/core.hpp

@@ -436,6 +436,12 @@ class Core {
   DATADOG_API bool Start();
   DATADOG_API void Stop();
 
+  /**
+   * FOR INTERNAL USE ONLY - This function is not part of the public API and may change
+   * without notice. Do not use in production code.
+   */
+  DATADOG_API void Internal_FlushWork();
+
  private:
   // Forbid copying/moving: we use std::shared_ptr<Core> at the API boundary
   Core(const Core&) = delete;

src/datadog/cpp/core.cpp

@@ -230,4 +230,10 @@ void Core::Stop() {
   }
 }
 
+void Core::Internal_FlushWork() {
+  if (_impl) {
+    _impl->FlushWork();
+  }
+}
+
 }  // namespace datadog

src/datadog/impl/core/core.cpp

@@ -7,7 +7,9 @@
 #include "datadog/impl/core/core.hpp"
 
 #include <algorithm>
+#include <future>
 #include <iostream>
+#include <memory>
 #include <sstream>
 
 #include "datadog/impl/core/attribute/merge.hpp"
@@ -609,6 +611,52 @@ void Core::Stop() {
   _state = CoreState::Initialized;
 }
 
+void Core::FlushWork() {
+  // Nothing to flush unless the Core is actually running: features cannot enqueue work
+  // on the context, messaging, or storage threads when the Core is not started, so any
+  // prior work is either already complete (from a previous run) or impossible (from
+  // this run).
+  if (_state != CoreState::Started) {
+    return;
+  }
+  DATADOG_ASSERT(_context_queue, "_context_queue is invalid while core is running");
+  DATADOG_ASSERT(_message_bus, "_message_bus is invalid while core is running");
+  DATADOG_ASSERT(_storage_queue, "_storage_queue is invalid while core is running");
+
+  // Drain the three queues in order: context thread → message bus → storage thread.
+  // The order matters: each FlushWork sentinel must be enqueued after the prior thread
+  // has finished producing work for the next queue.
+  //
+  // - The context-thread FlushWork sentinel waits for prior context-thread work to
+  //   finish; any storage writes (via `EventWriter`) and any messages broadcast (via
+  //   the `MessagePublisher`) by that work are enqueued in their respective queues by
+  //   the time it fulfills.
+  // - The message-bus FlushWork sentinel waits for prior messages to be handled; any
+  //   storage writes that message handlers triggered are enqueued in the storage queue
+  //   by the time it fulfills.
+  // - The storage-thread FlushWork sentinel waits for all preceding writes to be
+  //   flushed to disk.
+  //
+  // Precondition: message handlers do not enqueue further work on the context thread
+  // or the message bus. If a future handler needs to do so, this routine will need
+  // to drain those cascading queues as well.
+
+  auto ctx_done = std::make_shared<std::promise<void>>();
+  std::future<void> ctx_future = ctx_done->get_future();
+  _context_queue->Push([ctx_done]() { ctx_done->set_value(); });
+  ctx_future.wait();
+
+  auto msg_done = std::make_shared<std::promise<void>>();
+  std::future<void> msg_future = msg_done->get_future();
+  _message_bus->Send(FeatureMessage{FlushWorkMessage{msg_done}});
+  msg_future.wait();
+
+  auto stor_done = std::make_shared<std::promise<void>>();
+  std::future<void> stor_future = stor_done->get_future();
+  _storage_queue->Push(StorageMessage::FlushWork(stor_done));
+  stor_future.wait();
+}
+
 bool Core::EnqueueStorageWrite(
     FeatureId feature_id,
     Block event,

src/datadog/impl/core/core.hpp

@@ -386,6 +386,25 @@ class Core {
    */
   void Stop();
 
+  /**
+   * Blocks until all work previously-enqueued on the context thread has been processed,
+   * all messages those handlers broadcast on the internal message bus have been
+   * dispatched, and all writes those handlers triggered (whether enqueued on the
+   * storage thread or written synchronously via a message handler) have completed.
+   * Does not stop the SDK, and does not flush HTTP requests.
+   *
+   * Calling this function from the context, messaging, or storage threads would
+   * deadlock. This function is not used in production code; it is directly exposed for
+   * use by an internal-only API function. (The ability to flush pending work is
+   * provided solely for internal use in integration tests.)
+   *
+   * This function only flushes work that was already enqueued when it was called. If a
+   * message-bus handler reacts to a message by pushing new work onto the context thread
+   * or message bus, that new work is NOT covered, and the caller must call this
+   * function again to flush it. (No message handler currently does this.)
+   */
+  void FlushWork();
+
  private:
   bool EnqueueStorageWrite(
       FeatureId feature_id,

src/datadog/impl/core/feature_message.hpp

@@ -6,6 +6,8 @@
 
 #pragma once
 
+#include <future>
+#include <memory>
 #include <variant>
 
 #include "datadog/attribute.hpp"
@@ -95,6 +97,15 @@ struct CrashReportProcessedMessage {
   CrashReport crash;
 };
 
+/**
+ * Sentinel produced by `Core::FlushWork()` to synchronize with the messaging thread.
+ * Carries a shared promise that the messaging thread fulfills once it pops this message
+ * from the queue; not delivered to feature handlers.
+ */
+struct FlushWorkMessage {
+  std::shared_ptr<std::promise<void>> done;
+};
+
 /**
  * Discriminated union of all message types that can be dispatched through the
  * `MessageBus`. Add new variants here as additional cross-feature communication needs
@@ -106,6 +117,7 @@ using FeatureMessage = std::variant<
     RumActiveViewUpdatedMessage,
     RumActiveViewLostMessage,
     RumGlobalAttributesChangedMessage,
-    CrashReportProcessedMessage>;
+    CrashReportProcessedMessage,
+    FlushWorkMessage>;
 
 }  // namespace datadog::impl

src/datadog/impl/core/messaging_thread.cpp

@@ -12,6 +12,15 @@ void MessagingThreadMain(const DiagnosticLogger& diagnostic_logger, MessageBus&
   diagnostic_logger.Debug("Messaging thread starting");
 
   while (auto msg = bus._queue.Pop()) {
+    // FlushWork messages are not delivered to feature handlers; fulfill the carried
+    // promise and continue.
+    if (auto* barrier = std::get_if<FlushWorkMessage>(&*msg)) {
+      if (barrier->done) {
+        barrier->done->set_value();
+      }
+      continue;
+    }
+
     for (auto& handler : bus._handlers) {
       handler(*msg);
     }

src/datadog/impl/core/storage_queue.cpp

@@ -34,6 +34,10 @@ StorageMessage::~StorageMessage() {
     case StorageMessageType::EventGenerated:
       payload.event_generated.~StorageMessage_EventGenerated();
       break;
+
+    case StorageMessageType::FlushWork:
+      payload.flush.~StorageMessage_FlushWork();
+      break;
   }
 }
 
@@ -53,6 +57,10 @@ StorageMessage::StorageMessage(StorageMessage&& other) noexcept : type(other.typ
       new (&payload.event_generated)
           StorageMessage_EventGenerated(std::move(other.payload.event_generated));
       break;
+
+    case StorageMessageType::FlushWork:
+      new (&payload.flush) StorageMessage_FlushWork(std::move(other.payload.flush));
+      break;
   }
 }
 
@@ -77,6 +85,10 @@ StorageMessage& StorageMessage::operator=(StorageMessage&& other) noexcept {
         new (&payload.event_generated)
             StorageMessage_EventGenerated(std::move(other.payload.event_generated));
         break;
+
+      case StorageMessageType::FlushWork:
+        new (&payload.flush) StorageMessage_FlushWork(std::move(other.payload.flush));
+        break;
     }
   }
   return *this;
@@ -102,4 +114,10 @@ StorageMessage StorageMessage::EventGenerated(
   return m;
 }
 
+StorageMessage StorageMessage::FlushWork(std::shared_ptr<std::promise<void>> done) {
+  StorageMessage m{StorageMessageType::FlushWork};
+  new (&m.payload.flush) StorageMessage_FlushWork{std::move(done)};
+  return m;
+}
+
 }  // namespace datadog::impl

src/datadog/impl/core/storage_queue.hpp

@@ -7,6 +7,8 @@
 #pragma once
 
 #include <cinttypes>
+#include <future>
+#include <memory>
 #include <utility>
 #include <vector>
 
@@ -41,6 +43,7 @@ namespace datadog::impl {
 enum class StorageMessageType : uint8_t {
   TrackingConsentChanged,
   EventGenerated,
+  FlushWork,
 };
 
 /**
@@ -92,6 +95,15 @@ struct StorageMessage_EventGenerated {
   );
 };
 
+/**
+ * Sentinel used by `Core::FlushWork()` to wait for all prior writes to be flushed to
+ * disk. The storage thread fulfills `done` once it pops this message, unblocking the
+ * caller's future.
+ */
+struct StorageMessage_FlushWork {
+  std::shared_ptr<std::promise<void>> done;
+};
+
 /**
  * A message sent to the storage thread.
  */
@@ -111,6 +123,7 @@ struct StorageMessage {
   union Payload {
     StorageMessage_TrackingConsentChanged tracking_consent_changed;
     StorageMessage_EventGenerated event_generated;
+    StorageMessage_FlushWork flush;
 
     /**
      * Messages are initialized using static functions that use placement new to
@@ -163,6 +176,12 @@ struct StorageMessage {
       Block event_metadata,
       bool bypass_tracking_consent
   );
+
+  /**
+   * Creates a new FlushWork message that will unblock the caller's future once the
+   * storage thread pops it.
+   */
+  static StorageMessage FlushWork(std::shared_ptr<std::promise<void>> done);
 };
 
 /**

src/datadog/impl/core/storage_thread.cpp

@@ -115,6 +115,14 @@ void StorageThreadMain(
             diagnostic_logger, features, item->payload.event_generated
         );
         break;
+
+      case StorageMessageType::FlushWork:
+        // All prior writes have been drained from the queue by virtue of FIFO
+        // ordering; unblock the waiter
+        if (item->payload.flush.done) {
+          item->payload.flush.done->set_value();
+        }
+        break;
     }
   }