From 348b997286e9a9fc4b25f09a792106082d1a7176 Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Mon, 11 May 2026 13:26:54 -0400 Subject: [PATCH] fix: Add repl flush-work command refs: RUM-16267 This PR adds a repl command, `flush-work`, which calls an internal-only function in the C++ API, `Core::Internal_FlushWork()`, which in turn calls `impl::Core::FlushWork()`. This function blocks until all previously-enqueued context-thread work is completed, _including_ any event writes (handled on the storage thread) or feature messages (handled on the messaging thread) that were enqueued as the result of any such context-thread work. --- examples/repl/commands.hpp | 1 + examples/repl/commands_util.cpp | 10 +++ examples/repl/main.cpp | 3 + include-cpp/datadog/core.hpp | 6 ++ src/datadog/cpp/core.cpp | 6 ++ src/datadog/impl/core/core.cpp | 48 +++++++++++ src/datadog/impl/core/core.hpp | 19 +++++ src/datadog/impl/core/feature_message.hpp | 14 +++- src/datadog/impl/core/messaging_thread.cpp | 9 ++ src/datadog/impl/core/storage_queue.cpp | 18 ++++ src/datadog/impl/core/storage_queue.hpp | 19 +++++ src/datadog/impl/core/storage_thread.cpp | 8 ++ tests/impl/core/core_test.cpp | 91 +++++++++++++++++++++ tests/impl/core/storage_queue_test.cpp | 73 +++++++++++++++++ tests/impl/core/storage_thread_test.cpp | 48 +++++++++++ tools/integration-test/tests/crash_smoke.py | 2 +- 16 files changed, 373 insertions(+), 2 deletions(-) diff --git a/examples/repl/commands.hpp b/examples/repl/commands.hpp index c0f01b0e..065e4873 100644 --- a/examples/repl/commands.hpp +++ b/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); diff --git a/examples/repl/commands_util.cpp b/examples/repl/commands_util.cpp index 41f1c5ef..cb69e40b 100644 --- a/examples/repl/commands_util.cpp +++ b/examples/repl/commands_util.cpp @@ -9,6 +9,8 @@ #include #include +#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; diff --git a/examples/repl/main.cpp b/examples/repl/main.cpp index aa27c6df..5ffee7c8 100644 --- a/examples/repl/main.cpp +++ b/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()); } diff --git a/include-cpp/datadog/core.hpp b/include-cpp/datadog/core.hpp index 8f9cb467..f636b4be 100644 --- a/include-cpp/datadog/core.hpp +++ b/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 at the API boundary Core(const Core&) = delete; diff --git a/src/datadog/cpp/core.cpp b/src/datadog/cpp/core.cpp index 82641e51..58841d59 100644 --- a/src/datadog/cpp/core.cpp +++ b/src/datadog/cpp/core.cpp @@ -230,4 +230,10 @@ void Core::Stop() { } } +void Core::Internal_FlushWork() { + if (_impl) { + _impl->FlushWork(); + } +} + } // namespace datadog diff --git a/src/datadog/impl/core/core.cpp b/src/datadog/impl/core/core.cpp index 26e6fb56..ac239f50 100644 --- a/src/datadog/impl/core/core.cpp +++ b/src/datadog/impl/core/core.cpp @@ -7,7 +7,9 @@ #include "datadog/impl/core/core.hpp" #include +#include #include +#include #include #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::future 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::future msg_future = msg_done->get_future(); + _message_bus->Send(FeatureMessage{FlushWorkMessage{msg_done}}); + msg_future.wait(); + + auto stor_done = std::make_shared>(); + std::future stor_future = stor_done->get_future(); + _storage_queue->Push(StorageMessage::FlushWork(stor_done)); + stor_future.wait(); +} + bool Core::EnqueueStorageWrite( FeatureId feature_id, Block event, diff --git a/src/datadog/impl/core/core.hpp b/src/datadog/impl/core/core.hpp index 99b32466..b5e3351f 100644 --- a/src/datadog/impl/core/core.hpp +++ b/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, diff --git a/src/datadog/impl/core/feature_message.hpp b/src/datadog/impl/core/feature_message.hpp index e900f8a8..4f5df1ad 100644 --- a/src/datadog/impl/core/feature_message.hpp +++ b/src/datadog/impl/core/feature_message.hpp @@ -6,6 +6,8 @@ #pragma once +#include +#include #include #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> 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 diff --git a/src/datadog/impl/core/messaging_thread.cpp b/src/datadog/impl/core/messaging_thread.cpp index 741a93b6..a44deeb0 100644 --- a/src/datadog/impl/core/messaging_thread.cpp +++ b/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(&*msg)) { + if (barrier->done) { + barrier->done->set_value(); + } + continue; + } + for (auto& handler : bus._handlers) { handler(*msg); } diff --git a/src/datadog/impl/core/storage_queue.cpp b/src/datadog/impl/core/storage_queue.cpp index cf8471ac..e24c5a9f 100644 --- a/src/datadog/impl/core/storage_queue.cpp +++ b/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> done) { + StorageMessage m{StorageMessageType::FlushWork}; + new (&m.payload.flush) StorageMessage_FlushWork{std::move(done)}; + return m; +} + } // namespace datadog::impl diff --git a/src/datadog/impl/core/storage_queue.hpp b/src/datadog/impl/core/storage_queue.hpp index c19a89b6..dd8f1d55 100644 --- a/src/datadog/impl/core/storage_queue.hpp +++ b/src/datadog/impl/core/storage_queue.hpp @@ -7,6 +7,8 @@ #pragma once #include +#include +#include #include #include @@ -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> 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> done); }; /** diff --git a/src/datadog/impl/core/storage_thread.cpp b/src/datadog/impl/core/storage_thread.cpp index 693f0f77..f34b004d 100644 --- a/src/datadog/impl/core/storage_thread.cpp +++ b/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; } } diff --git a/tests/impl/core/core_test.cpp b/tests/impl/core/core_test.cpp index f87f5aeb..a943473f 100644 --- a/tests/impl/core/core_test.cpp +++ b/tests/impl/core/core_test.cpp @@ -7,6 +7,7 @@ #include "datadog/impl/core/core.hpp" #include +#include #include #include "datadog/impl/core/feature_message.hpp" @@ -16,6 +17,7 @@ #include "mock/filesystem.hpp" #include "mock/http_client.hpp" #include "mock/system_info.hpp" +#include "mock/tlv.hpp" #include "support/catch.hpp" #include "support/core.hpp" #include "support/diagnostics.hpp" @@ -446,3 +448,92 @@ TEST_CASE("Core Messaging", "[unit]") { } } } + +TEST_CASE("Core::FlushWork", "[unit]") { + SECTION("M flush a pending write to disk W called on a running core") { + // Given a running core with a registered feature, frozen at a known timestamp so + // that batch filenames are predictable + CoreTestHarness test = CoreTestHarness::Init(/*flush_http_requests=*/false); + test.clock.FreezeAtMilliseconds(1708675309000); + auto feature = std::make_shared(); + REQUIRE(test.core.RegisterFeature(feature)); + REQUIRE(test.core.Start()); + + // And no events have reached the granted-consent storage directory yet + REQUIRE(test.fs.Ls("app/.datadog/main/12345/testfeature/v1").empty()); + + // When we generate an event and then call FlushWork() + REQUIRE(feature->GenerateEvent("event-0")); + test.core.FlushWork(); + + // Then the event has been written to disk before FlushWork() returned, without + // needing to stop the core + auto names = test.fs.Ls("app/.datadog/main/12345/testfeature/v1"); + REQUIRE(names.size() == 1); + REQUIRE( + test.fs.Cat("app/.datadog/main/12345/testfeature/v1/" + names.front()) == + MockTLVFile().AppendEvent("event-0").ToString() + ); + + // Clean up + test.core.Stop(); + } + + SECTION("M return immediately W core is not started") { + // Given a core that has been initialized but never started + DiagnosticMessageBuffer diagnostics; + impl::Core core = _make_core(diagnostics); + REQUIRE(core.Init()); + + // When FlushWork() is called, it returns without hanging (no context or storage + // threads exist to enqueue work to) + const auto start = std::chrono::steady_clock::now(); + core.FlushWork(); + const auto elapsed = std::chrono::steady_clock::now() - start; + REQUIRE(elapsed < std::chrono::seconds(1)); + } + + SECTION("M return immediately W core has been stopped") { + // Given a core that was started and then stopped + CoreTestHarness test = CoreTestHarness::Init(/*flush_http_requests=*/false); + auto feature = std::make_shared(); + REQUIRE(test.core.RegisterFeature(feature)); + REQUIRE(test.core.Start()); + test.core.Stop(); + + // When FlushWork() is called after Stop(), it returns without hanging + const auto start = std::chrono::steady_clock::now(); + test.core.FlushWork(); + const auto elapsed = std::chrono::steady_clock::now() - start; + REQUIRE(elapsed < std::chrono::seconds(1)); + } + + SECTION("M drain the message bus W messages were broadcast before FlushWork") { + // Given a running core with a feature that subscribes to ContextChangedMessage + CoreTestHarness test = CoreTestHarness::Init(/*flush_http_requests=*/false); + auto feature = std::make_shared(); + REQUIRE(test.core.RegisterFeature(feature)); + + // When the core starts, two ContextChangedMessages are broadcast: one from + // Core::Start()'s no-op UpdateContext, and another from + // MessageHandlerFeature::Start() which enqueues its own UpdateContext on the + // context thread. Neither has been delivered to the handler yet at this point. + REQUIRE(test.core.Start()); + + // When FlushWork() is called, both messages are delivered to the handler + // before it returns (no Stop() required to drain the bus) + test.core.FlushWork(); + REQUIRE(feature->messages_received.load() == 2); + + // And: when an additional context change is triggered from the main thread, + // followed by another FlushWork(), the third message is also delivered + test.core.SetTrackingConsent(TrackingConsent::Pending); + test.core.FlushWork(); + REQUIRE(feature->messages_received.load() == 3); + REQUIRE(feature->final_context.has_value()); + REQUIRE(feature->final_context->tracking_consent == TrackingConsent::Pending); + + // Clean up + test.core.Stop(); + } +} diff --git a/tests/impl/core/storage_queue_test.cpp b/tests/impl/core/storage_queue_test.cpp index 3b829f44..ce2d4657 100644 --- a/tests/impl/core/storage_queue_test.cpp +++ b/tests/impl/core/storage_queue_test.cpp @@ -7,6 +7,7 @@ #include "datadog/impl/core/storage_queue.hpp" #include +#include #include #include @@ -164,4 +165,76 @@ TEST_CASE("StorageMessage", "[unit]") { } // Destructor is called here - test passes if no crashes occur } + + SECTION("M create FlushWork message W static factory method used") { + // Given a shared promise that the storage thread will fulfill + auto promise = std::make_shared>(); + const auto* raw = promise.get(); + + // When creating a FlushWork message via the factory + auto message = StorageMessage::FlushWork(promise); + + // Then the message has the correct type and the payload owns the same promise + REQUIRE(message.type == StorageMessageType::FlushWork); + REQUIRE(message.payload.flush.done.get() == raw); + } + + SECTION("M move construct properly W union contains FlushWork") { + // Given a FlushWork message carrying a shared promise, and a future obtained from + // that promise before the move + auto promise = std::make_shared>(); + std::future future = promise->get_future(); + auto original = StorageMessage::FlushWork(std::move(promise)); + + // When move constructing from it + auto moved = std::move(original); + + // Then the moved message retains the type and owns the promise + REQUIRE(moved.type == StorageMessageType::FlushWork); + REQUIRE(moved.payload.flush.done != nullptr); + + // And fulfilling via the moved-into message unblocks the caller's future + moved.payload.flush.done->set_value(); + REQUIRE(future.wait_for(std::chrono::seconds(0)) == std::future_status::ready); + } + + SECTION( + "M move assign properly W source is FlushWork and destination is EventGenerated" + ) { + // Given a destination EventGenerated message and a source FlushWork message; + // capture a future from the FlushWork's promise before the move + FeatureId feature_id = CreateFeatureId("SWAP"); + Block event_data{"to be overwritten"}; + auto destination = + StorageMessage::EventGenerated(feature_id, event_data, {}, false); + auto promise = std::make_shared>(); + std::future future = promise->get_future(); + auto source = StorageMessage::FlushWork(std::move(promise)); + + // When move-assigning across types + destination = std::move(source); + + // Then the destination correctly adopts the FlushWork type and promise + REQUIRE(destination.type == StorageMessageType::FlushWork); + REQUIRE(destination.payload.flush.done != nullptr); + + // And the carried promise still drives the original future + destination.payload.flush.done->set_value(); + REQUIRE(future.wait_for(std::chrono::seconds(0)) == std::future_status::ready); + } + + SECTION("M properly destruct W union contains FlushWork") { + // Given a FlushWork message in a limited scope; capture a future before destruction + auto promise = std::make_shared>(); + std::future future = promise->get_future(); + { + auto message = StorageMessage::FlushWork(promise); + REQUIRE(message.type == StorageMessageType::FlushWork); + } + // Destruction releases the message's reference to the promise; this thread's + // shared_ptr (`promise`) still owns it, so the future state is intact + REQUIRE(future.wait_for(std::chrono::seconds(0)) == std::future_status::timeout); + promise->set_value(); + REQUIRE(future.wait_for(std::chrono::seconds(0)) == std::future_status::ready); + } } diff --git a/tests/impl/core/storage_thread_test.cpp b/tests/impl/core/storage_thread_test.cpp index 15827a99..8653bec1 100644 --- a/tests/impl/core/storage_thread_test.cpp +++ b/tests/impl/core/storage_thread_test.cpp @@ -7,6 +7,8 @@ #include "datadog/impl/core/storage_thread.hpp" #include +#include +#include #include #include @@ -168,4 +170,50 @@ TEST_CASE("StorageThreadMain", "[unit]") { MockTLVFile().AppendEvent("bravo-0").ToString() ); } + + SECTION( + "M fulfill FlushWork sentinel after draining prior writes W FlushWork is queued" + ) { + // Given both features configured with Granted consent so writes land on disk + auto features = init_features(TrackingConsent::Granted, TrackingConsent::Granted); + REQUIRE(features.size() == 2); + + // And a queue containing an EventGenerated message followed by a FlushWork sentinel + // carrying a shared promise; capture the future before pushing + auto done = std::make_shared>(); + std::future future = done->get_future(); + StorageQueue queue; + REQUIRE(queue.Push( + StorageMessage::EventGenerated(CreateFeatureId("ALFA"), "alpha-0", {}, false) + )); + REQUIRE(queue.Push(StorageMessage::FlushWork(done))); + + // When the storage thread drains the queue + queue.Stop(); + StorageThreadMain(DiagnosticLogger{}, queue, features); + + // Then the prior write has reached 'alpha/v1' on disk, and the FlushWork sentinel + // has fulfilled the promise (FIFO ordering: the sentinel can only have been + // processed after all preceding writes) + auto alpha_granted_files = fs.Ls(alpha_prefix + "v1"); + REQUIRE(alpha_granted_files.size() == 1); + REQUIRE( + fs.Cat(alpha_prefix + "v1/" + alpha_granted_files.front()) == + MockTLVFile().AppendEvent("alpha-0").ToString() + ); + REQUIRE(future.wait_for(std::chrono::seconds(0)) == std::future_status::ready); + } + + SECTION("M tolerate FlushWork message with null promise W done is unset") { + // Given a FlushWork message whose `done` shared_ptr is null + auto features = init_features(TrackingConsent::Granted, TrackingConsent::Granted); + StorageQueue queue; + REQUIRE(queue.Push(StorageMessage::FlushWork(nullptr))); + + // When the storage thread drains the queue + queue.Stop(); + + // Then it processes the message without crashing + StorageThreadMain(DiagnosticLogger{}, queue, features); + } } diff --git a/tools/integration-test/tests/crash_smoke.py b/tools/integration-test/tests/crash_smoke.py index d2377ac1..069c1b2f 100644 --- a/tools/integration-test/tests/crash_smoke.py +++ b/tools/integration-test/tests/crash_smoke.py @@ -24,7 +24,7 @@ async def main(t: TestContext): register-rum start-core start-view foo - sleep 10 + flush-work crash bad-sdk-usage """) await p1.join()