From f11c03395ebf25259c0e10303ff66ea5c354688c Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Mon, 31 Aug 2026 14:30:30 -0700 Subject: [PATCH 1/3] test: Add repro for FDv1 adapter use-after-free on close --- .../tests/fdv1_adapter_synchronizer_test.cpp | 52 +++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/libs/server-sdk/tests/fdv1_adapter_synchronizer_test.cpp b/libs/server-sdk/tests/fdv1_adapter_synchronizer_test.cpp index 26de674b6..a4992aefd 100644 --- a/libs/server-sdk/tests/fdv1_adapter_synchronizer_test.cpp +++ b/libs/server-sdk/tests/fdv1_adapter_synchronizer_test.cpp @@ -65,8 +65,60 @@ FDv1AdapterSynchronizer::SourceBuilder MakeMockBuilder( }; } +// FDv1 source that captures the destination pointer and never completes its +// ShutdownAsync. It models a real StreamingDataSource or PollingDataSource with +// a callback still in flight during teardown. The source keeps the destination +// pointer. It has not yet signaled that shutdown is complete. +class DeferredShutdownFDv1Source final : public IDataSynchronizer { + public: + void StartAsync(IDestination* destination, + data_model::SDKDataSet const* /*bootstrap*/) override { + destination_ = destination; + } + + // Keeps the completion but never invokes it, so the adapter never learns + // that in-flight work has drained. + void ShutdownAsync(std::function completion) override { + completion_ = std::move(completion); + } + + std::string const& Identity() const override { + static std::string const id = "deferred fdv1"; + return id; + } + + IDestination* destination_ = nullptr; + std::function completion_; +}; + } // namespace +// The IDataSynchronizer contract states the destination pointer stays valid +// until the ShutdownAsync completion handler is called. Close() fires a no-op +// completion and returns without waiting. The destructor then frees the +// destination. A source callback still in flight then lands on freed memory. +// AddressSanitizer reports the heap-use-after-free. +TEST(FDv1AdapterSynchronizerTest, + SourceCallbackAfterCloseHitsFreedDestination) { + // Outlives the adapter, like a real source whose in-flight callback holds a + // shared_from_this reference across the adapter's teardown. + auto source = std::make_shared(); + { + FDv1AdapterSynchronizer adapter( + [source](DataSourceStatusManager&) { return source; }); + adapter.Next(data_model::Selector{}); // triggers StartAsync + } // adapter destroyed: Close() runs, then the destination is freed + + // StartAsync ran and handed the destination to the source. + ASSERT_NE(source->destination_, nullptr); + + // The source delivers its in-flight callback into the freed destination. + data_model::Flag flag; + flag.key = "late"; + flag.version = 1; + source->destination_->Upsert("late", data_model::FlagDescriptor(flag)); +} + TEST(FDv1AdapterSynchronizerTest, FirstNextStartsFDv1Source) { MockFDv1Source* source = nullptr; FDv1AdapterSynchronizer adapter(MakeMockBuilder(&source)); From 33f52fc0f2101067f7b9538d7d2d0b48d12595b4 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Mon, 31 Aug 2026 16:05:07 -0700 Subject: [PATCH 2/3] test: Model a conforming FDv1 source in the adapter shutdown test --- .../tests/fdv1_adapter_synchronizer_test.cpp | 75 ++++++++++++------- 1 file changed, 49 insertions(+), 26 deletions(-) diff --git a/libs/server-sdk/tests/fdv1_adapter_synchronizer_test.cpp b/libs/server-sdk/tests/fdv1_adapter_synchronizer_test.cpp index a4992aefd..537afe0c7 100644 --- a/libs/server-sdk/tests/fdv1_adapter_synchronizer_test.cpp +++ b/libs/server-sdk/tests/fdv1_adapter_synchronizer_test.cpp @@ -2,6 +2,10 @@ #include +#include +#include +#include + #include #include #include @@ -65,58 +69,77 @@ FDv1AdapterSynchronizer::SourceBuilder MakeMockBuilder( }; } -// FDv1 source that captures the destination pointer and never completes its -// ShutdownAsync. It models a real StreamingDataSource or PollingDataSource with -// a callback still in flight during teardown. The source keeps the destination -// pointer. It has not yet signaled that shutdown is complete. -class DeferredShutdownFDv1Source final : public IDataSynchronizer { +// FDv1 source that mirrors the real StreamingDataSource/PollingDataSource +// teardown. It keeps itself alive with shared_from_this, defers its +// ShutdownAsync completion, and delivers one more callback to the destination +// before signaling completion. The contract requires the destination to stay +// valid until then. +class DeferredCompletionFDv1Source final + : public IDataSynchronizer, + public std::enable_shared_from_this { public: + explicit DeferredCompletionFDv1Source(boost::asio::any_io_executor executor) + : executor_(std::move(executor)) {} + void StartAsync(IDestination* destination, data_model::SDKDataSet const* /*bootstrap*/) override { destination_ = destination; } - // Keeps the completion but never invokes it, so the adapter never learns - // that in-flight work has drained. + // Posts the drain: one in-flight upsert into the destination, then the + // completion. Holds shared_from_this so the source outlives the adapter's + // teardown, as the real sources do during async shutdown. void ShutdownAsync(std::function completion) override { - completion_ = std::move(completion); + boost::asio::post(executor_, [self = shared_from_this(), + completion = std::move(completion)]() { + data_model::Flag flag; + flag.key = "late"; + flag.version = 1; + self->destination_->Upsert("late", + data_model::FlagDescriptor(flag)); + self->completion_invoked = true; + if (completion) { + completion(); + } + }); } std::string const& Identity() const override { - static std::string const id = "deferred fdv1"; + static std::string const id = "deferred completion fdv1"; return id; } + boost::asio::any_io_executor executor_; IDestination* destination_ = nullptr; - std::function completion_; + bool completion_invoked = false; }; } // namespace // The IDataSynchronizer contract states the destination pointer stays valid -// until the ShutdownAsync completion handler is called. Close() fires a no-op -// completion and returns without waiting. The destructor then frees the -// destination. A source callback still in flight then lands on freed memory. -// AddressSanitizer reports the heap-use-after-free. -TEST(FDv1AdapterSynchronizerTest, - SourceCallbackAfterCloseHitsFreedDestination) { - // Outlives the adapter, like a real source whose in-flight callback holds a - // shared_from_this reference across the adapter's teardown. - auto source = std::make_shared(); +// until the ShutdownAsync completion handler is called. This source upserts +// once during shutdown and then fires the completion. The adapter must keep the +// destination alive until the completion fires. Otherwise the drain lands on +// freed memory, a use-after-free. +TEST(FDv1AdapterSynchronizerTest, DestinationStaysValidUntilShutdownCompletes) { + boost::asio::io_context ioc; + auto source = + std::make_shared(ioc.get_executor()); { FDv1AdapterSynchronizer adapter( [source](DataSourceStatusManager&) { return source; }); adapter.Next(data_model::Selector{}); // triggers StartAsync - } // adapter destroyed: Close() runs, then the destination is freed + } // adapter destroyed: Close() requests shutdown, the drain is still + // pending - // StartAsync ran and handed the destination to the source. + // StartAsync handed the destination to the source. ASSERT_NE(source->destination_, nullptr); - // The source delivers its in-flight callback into the freed destination. - data_model::Flag flag; - flag.key = "late"; - flag.version = 1; - source->destination_->Upsert("late", data_model::FlagDescriptor(flag)); + // Run the pending drain: the source upserts into the destination, then + // fires the shutdown completion. + ioc.run(); + + EXPECT_TRUE(source->completion_invoked); } TEST(FDv1AdapterSynchronizerTest, FirstNextStartsFDv1Source) { From 2ec992c5a060205103d203069b9128b7c194d483 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Tue, 1 Sep 2026 12:52:00 -0700 Subject: [PATCH 3/3] fix: Hand FDv1 source destination and status manager as smart pointers --- libs/server-sdk/src/client_impl.cpp | 17 +- libs/server-sdk/src/client_impl.hpp | 2 +- .../source/idata_synchronizer.hpp | 6 +- .../background_sync_system.cpp | 9 +- .../background_sync_system.hpp | 5 +- .../sources/polling/polling_data_source.cpp | 37 ++-- .../sources/polling/polling_data_source.hpp | 15 +- .../sources/streaming/event_handler.cpp | 38 ++-- .../sources/streaming/event_handler.hpp | 10 +- .../streaming/streaming_data_source.cpp | 32 ++-- .../streaming/streaming_data_source.hpp | 9 +- .../fdv2/fdv1_adapter_synchronizer.cpp | 8 +- .../fdv2/fdv1_adapter_synchronizer.hpp | 14 +- .../fdv2/synchronizer_factories.cpp | 14 +- .../tests/data_source_event_handler_test.cpp | 166 +++++++++--------- .../tests/fdv1_adapter_synchronizer_test.cpp | 23 +-- 16 files changed, 213 insertions(+), 192 deletions(-) diff --git a/libs/server-sdk/src/client_impl.cpp b/libs/server-sdk/src/client_impl.cpp index ba04dda7a..5973619e6 100644 --- a/libs/server-sdk/src/client_impl.cpp +++ b/libs/server-sdk/src/client_impl.cpp @@ -70,7 +70,7 @@ static std::unique_ptr MakeBackgroundSyncSystem( config::built::BackgroundSyncConfig const& cfg, config::built::HttpProperties const& http_properties, boost::asio::any_io_executor const& executor, - data_components::DataSourceStatusManager& status_manager, + std::shared_ptr status_manager, Logger& logger) { return std::make_unique( endpoints, cfg, http_properties, executor, status_manager, logger); @@ -161,10 +161,10 @@ static std::unique_ptr MakeDataSystem( config::built::HttpProperties const& http_properties, Config const& config, boost::asio::any_io_executor const& executor, - data_components::DataSourceStatusManager& status_manager, + std::shared_ptr status_manager, Logger& logger) { if (config.DataSystemConfig().disabled) { - return std::make_unique(status_manager); + return std::make_unique(*status_manager); } auto data_source_properties = @@ -178,12 +178,12 @@ static std::unique_ptr MakeDataSystem( executor, status_manager, logger); }, [&](config::built::LazyLoadConfig const& cfg) { - return MakeLazyLoadSystem(cfg, status_manager, logger); + return MakeLazyLoadSystem(cfg, *status_manager, logger); }, [&](config::built::FDv2Config const& cfg) { return MakeFDv2System(config.ServiceEndpoints(), cfg, data_source_properties, executor, - status_manager, logger); + *status_manager, logger); }, }, config.DataSystemConfig().system_); @@ -238,7 +238,8 @@ ClientImpl::ClientImpl(Config config, std::string const& version) logger_(MakeLogger(config.Logging())), ioc_(kAsioConcurrencyHint), work_(boost::asio::make_work_guard(ioc_)), - status_manager_(), + status_manager_( + std::make_shared()), data_system_(MakeDataSystem(http_properties_, config_, ioc_.get_executor(), @@ -290,7 +291,7 @@ std::future ClientImpl::StartAsync() { auto pr = std::make_shared>(); auto fut = pr->get_future(); - status_manager_.OnDataSourceStatusChangeEx([this, pr](auto _) { + status_manager_->OnDataSourceStatusChangeEx([this, pr](auto _) { if (data_system_->Initialized()) { pr->set_value(true); return true; /* delete this change listener since the @@ -778,7 +779,7 @@ Value ClientImpl::JsonVariation(Context const& ctx, } IDataSourceStatusProvider& ClientImpl::DataSourceStatus() { - return status_manager_; + return *status_manager_; } IBigSegmentStoreStatusProvider& ClientImpl::BigSegmentStoreStatus() { diff --git a/libs/server-sdk/src/client_impl.hpp b/libs/server-sdk/src/client_impl.hpp index 8466a682f..3af41c46f 100644 --- a/libs/server-sdk/src/client_impl.hpp +++ b/libs/server-sdk/src/client_impl.hpp @@ -250,7 +250,7 @@ class ClientImpl : public IClient { boost::asio::executor_work_guard work_; - data_components::DataSourceStatusManager status_manager_; + std::shared_ptr status_manager_; // This is the main polymorphic component that constitutes the // guts of how data is retrieved (polling, streaming, persistent stores, diff --git a/libs/server-sdk/src/data_interfaces/source/idata_synchronizer.hpp b/libs/server-sdk/src/data_interfaces/source/idata_synchronizer.hpp index ea1250dfb..2491ec761 100644 --- a/libs/server-sdk/src/data_interfaces/source/idata_synchronizer.hpp +++ b/libs/server-sdk/src/data_interfaces/source/idata_synchronizer.hpp @@ -3,6 +3,7 @@ #include #include +#include #include #include @@ -30,12 +31,11 @@ class IDataSynchronizer { * The data may be used to optimize the synchronization process, e.g. by * obtaining a diff rather than a full dataset. * - * @param destination The destination to synchronize data into. Pointer is - * invalid after the ShutdownAsync completion handler is called. + * @param destination The destination to synchronize data into. * @param bootstrap_data Optional bootstrap data. * Pointer is valid only for this call. */ - virtual void StartAsync(IDestination* destination, + virtual void StartAsync(std::shared_ptr destination, data_model::SDKDataSet const* bootstrap_data) = 0; /** diff --git a/libs/server-sdk/src/data_systems/background_sync/background_sync_system.cpp b/libs/server-sdk/src/data_systems/background_sync/background_sync_system.cpp index 1a6de5c0a..85f5df495 100644 --- a/libs/server-sdk/src/data_systems/background_sync/background_sync_system.cpp +++ b/libs/server-sdk/src/data_systems/background_sync/background_sync_system.cpp @@ -10,9 +10,12 @@ BackgroundSync::BackgroundSync( config::built::BackgroundSyncConfig const& background_sync_config, config::built::HttpProperties http_properties, boost::asio::any_io_executor ioc, - data_components::DataSourceStatusManager& status_manager, + std::shared_ptr status_manager, Logger const& logger) - : store_(), change_notifier_(store_, store_), synchronizer_() { + : store_(), + change_notifier_( + std::make_shared(store_, store_)), + synchronizer_() { std::visit( [&](auto&& method_config) { using T = std::decay_t; @@ -34,7 +37,7 @@ BackgroundSync::BackgroundSync( } void BackgroundSync::Initialize() { - synchronizer_->StartAsync(&change_notifier_, + synchronizer_->StartAsync(change_notifier_, nullptr /* no bootstrap data supported yet */); } diff --git a/libs/server-sdk/src/data_systems/background_sync/background_sync_system.hpp b/libs/server-sdk/src/data_systems/background_sync/background_sync_system.hpp index a302b7d0d..5a91207ae 100644 --- a/libs/server-sdk/src/data_systems/background_sync/background_sync_system.hpp +++ b/libs/server-sdk/src/data_systems/background_sync/background_sync_system.hpp @@ -31,7 +31,8 @@ class BackgroundSync final : public data_interfaces::IDataSystem { config::built::BackgroundSyncConfig const& background_sync_config, config::built::HttpProperties http_properties, boost::asio::any_io_executor ioc, - data_components::DataSourceStatusManager& status_manager, + std::shared_ptr + status_manager, Logger const& logger); BackgroundSync(BackgroundSync const& item) = delete; @@ -57,7 +58,7 @@ class BackgroundSync final : public data_interfaces::IDataSystem { private: data_components::MemoryStore store_; - data_components::ChangeNotifier change_notifier_; + std::shared_ptr change_notifier_; // Needs to be shared to that the source can keep itself alive through // async operations. std::shared_ptr synchronizer_; diff --git a/libs/server-sdk/src/data_systems/background_sync/sources/polling/polling_data_source.cpp b/libs/server-sdk/src/data_systems/background_sync/sources/polling/polling_data_source.cpp index abb2a5478..f8569b3f6 100644 --- a/libs/server-sdk/src/data_systems/background_sync/sources/polling/polling_data_source.cpp +++ b/libs/server-sdk/src/data_systems/background_sync/sources/polling/polling_data_source.cpp @@ -3,8 +3,8 @@ #include #include -#include #include +#include #include #include @@ -65,19 +65,18 @@ std::string const& PollingDataSource::Identity() const { PollingDataSource::PollingDataSource( boost::asio::any_io_executor const& ioc, Logger const& logger, - data_components::DataSourceStatusManager& status_manager, + std::shared_ptr status_manager, config::built::ServiceEndpoints const& endpoints, config::built::BackgroundSyncConfig::PollingConfig const& data_source_config, config::built::HttpProperties const& http_properties) : logger_(logger), - status_manager_(status_manager), + status_manager_(std::move(status_manager)), requester_(ioc, http_properties.Tls()), polling_interval_(data_source_config.poll_interval), request_( MakeRequest(logger_, data_source_config, endpoints, http_properties)), - timer_(ioc), - sink_(nullptr) { + timer_(ioc) { if (polling_interval_ < data_source_config.min_polling_interval) { LD_LOG(logger_, LogLevel::kWarn) << "Polling interval too frequent, defaulting to " @@ -127,7 +126,7 @@ void PollingDataSource::HandlePollResult(network::HttpResult const& res) { if (res.IsError()) { auto const& error_message = res.ErrorMessage(); - status_manager_.SetState( + status_manager_->SetState( DataSourceStatus::DataSourceState::kInterrupted, DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError, error_message.has_value() ? *error_message : "unknown error"); @@ -141,7 +140,7 @@ void PollingDataSource::HandlePollResult(network::HttpResult const& res) { auto parsed = boost::json::parse(body.value(), error_code); if (error_code) { LD_LOG(logger_, LogLevel::kError) << kErrorParsingPut; - status_manager_.SetError( + status_manager_->SetError( DataSourceStatus::ErrorInfo::ErrorKind::kInvalidData, kErrorParsingPut); return; @@ -150,18 +149,20 @@ void PollingDataSource::HandlePollResult(network::HttpResult const& res) { tl::expected>(parsed); if (poll_result.has_value()) { - sink_->Init(std::move(*poll_result)); - status_manager_.SetState( - DataSourceStatus::DataSourceState::kValid); + if (auto sink = sink_.lock()) { + sink->Init(std::move(*poll_result)); + status_manager_->SetState( + DataSourceStatus::DataSourceState::kValid); + } return; } LD_LOG(logger_, LogLevel::kError) << kErrorPutInvalid; - status_manager_.SetError( + status_manager_->SetError( DataSourceStatus::ErrorInfo::ErrorKind::kInvalidData, kErrorPutInvalid); return; } - status_manager_.SetState( + status_manager_->SetState( DataSourceStatus::DataSourceState::kInterrupted, DataSourceStatus::ErrorInfo::ErrorKind::kUnknown, "polling response contained no body."); @@ -172,12 +173,12 @@ void PollingDataSource::HandlePollResult(network::HttpResult const& res) { // parse the body. } else { if (network::IsRecoverableStatus(res.Status())) { - status_manager_.SetState( + status_manager_->SetState( DataSourceStatus::DataSourceState::kInterrupted, res.Status(), launchdarkly::network::ErrorForStatusCode( res.Status(), "polling request", "will retry")); } else { - status_manager_.SetState( + status_manager_->SetState( DataSourceStatus::DataSourceState::kOff, res.Status(), launchdarkly::network::ErrorForStatusCode( res.Status(), "polling request", std::nullopt)); @@ -226,16 +227,16 @@ void PollingDataSource::StartPollingTimer() { } void PollingDataSource::StartAsync( - data_interfaces::IDestination* dest, + std::shared_ptr dest, data_model::SDKDataSet const* bootstrap_data) { boost::ignore_unused(bootstrap_data); sink_ = dest; - status_manager_.SetState(DataSourceStatus::DataSourceState::kInitializing); + status_manager_->SetState(DataSourceStatus::DataSourceState::kInitializing); if (!request_.Valid()) { LD_LOG(logger_, LogLevel::kError) << kCouldNotParseEndpoint; - status_manager_.SetState( + status_manager_->SetState( DataSourceStatus::DataSourceState::kOff, DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError, kCouldNotParseEndpoint); @@ -248,7 +249,7 @@ void PollingDataSource::StartAsync( } void PollingDataSource::ShutdownAsync(std::function completion) { - status_manager_.SetState(DataSourceStatus::DataSourceState::kInitializing); + status_manager_->SetState(DataSourceStatus::DataSourceState::kInitializing); timer_.cancel(); if (completion) { boost::asio::post(timer_.get_executor(), completion); diff --git a/libs/server-sdk/src/data_systems/background_sync/sources/polling/polling_data_source.hpp b/libs/server-sdk/src/data_systems/background_sync/sources/polling/polling_data_source.hpp index b9492c755..8294801d3 100644 --- a/libs/server-sdk/src/data_systems/background_sync/sources/polling/polling_data_source.hpp +++ b/libs/server-sdk/src/data_systems/background_sync/sources/polling/polling_data_source.hpp @@ -12,6 +12,7 @@ #include #include +#include namespace launchdarkly::server_side::data_systems { @@ -21,13 +22,14 @@ class PollingDataSource public: PollingDataSource(boost::asio::any_io_executor const& ioc, Logger const& logger, - data_components::DataSourceStatusManager& status_manager, + std::shared_ptr + status_manager, config::built::ServiceEndpoints const& endpoints, config::built::BackgroundSyncConfig::PollingConfig const& data_source_config, config::built::HttpProperties const& http_properties); - void StartAsync(data_interfaces::IDestination* dest, + void StartAsync(std::shared_ptr dest, data_model::SDKDataSet const* bootstrap_data) override; void ShutdownAsync(std::function completion) override; @@ -40,11 +42,8 @@ class PollingDataSource Logger const& logger_; - // Status manager is used to report the status of the data source. It must - // outlive the source. This source performs asynchronous - // operations, so a completion handler might invoke the status manager after - // it has been destroyed. - data_components::DataSourceStatusManager& status_manager_; + // Reports the status of the data source. + std::shared_ptr status_manager_; // Responsible for performing HTTP requests using boost::asio. network::AsioRequester requester_; @@ -66,7 +65,7 @@ class PollingDataSource std::chrono::time_point last_poll_start_; // Destination for all data obtained via polling. - data_interfaces::IDestination* sink_; + std::weak_ptr sink_; void StartPollingTimer(); }; diff --git a/libs/server-sdk/src/data_systems/background_sync/sources/streaming/event_handler.cpp b/libs/server-sdk/src/data_systems/background_sync/sources/streaming/event_handler.cpp index 078f00ca8..2ba56d6e3 100644 --- a/libs/server-sdk/src/data_systems/background_sync/sources/streaming/event_handler.cpp +++ b/libs/server-sdk/src/data_systems/background_sync/sources/streaming/event_handler.cpp @@ -130,20 +130,26 @@ static tl::expected tag_invoke( } DataSourceEventHandler::DataSourceEventHandler( - data_interfaces::IDestination& handler, + std::weak_ptr handler, Logger const& logger, - data_components::DataSourceStatusManager& status_manager) - : handler_(handler), logger_(logger), status_manager_(status_manager) {} + std::shared_ptr status_manager) + : handler_(std::move(handler)), + logger_(logger), + status_manager_(std::move(status_manager)) {} DataSourceEventHandler::MessageStatus DataSourceEventHandler::HandleMessage( std::string const& type, std::string const& data) { + auto handler = handler_.lock(); + if (!handler) { + return MessageStatus::kMessageHandled; + } if (type == "put") { boost::system::error_code error_code; auto parsed = boost::json::parse(data, error_code); if (error_code) { LD_LOG(logger_, LogLevel::kError) << kErrorParsingPut; - status_manager_.SetError( + status_manager_->SetError( DataSourceStatus::ErrorInfo::ErrorKind::kInvalidData, kErrorParsingPut); return MessageStatus::kInvalidMessage; @@ -154,7 +160,7 @@ DataSourceEventHandler::MessageStatus DataSourceEventHandler::HandleMessage( if (!res) { LD_LOG(logger_, LogLevel::kError) << kErrorPutInvalid; - status_manager_.SetError( + status_manager_->SetError( DataSourceStatus::ErrorInfo::ErrorKind::kInvalidData, kErrorPutInvalid); return MessageStatus::kInvalidMessage; @@ -162,8 +168,9 @@ DataSourceEventHandler::MessageStatus DataSourceEventHandler::HandleMessage( // Check the inner optional. if (res->has_value()) { - handler_.Init(std::move((*res)->data)); - status_manager_.SetState(DataSourceStatus::DataSourceState::kValid); + handler->Init(std::move((*res)->data)); + status_manager_->SetState( + DataSourceStatus::DataSourceState::kValid); return MessageStatus::kMessageHandled; } return MessageStatus::kMessageHandled; @@ -173,7 +180,7 @@ DataSourceEventHandler::MessageStatus DataSourceEventHandler::HandleMessage( auto parsed = boost::json::parse(data, error_code); if (error_code) { LD_LOG(logger_, LogLevel::kError) << kErrorParsingPut; - status_manager_.SetError( + status_manager_->SetError( DataSourceStatus::ErrorInfo::ErrorKind::kInvalidData, kErrorParsingPatch); return MessageStatus::kInvalidMessage; @@ -183,7 +190,7 @@ DataSourceEventHandler::MessageStatus DataSourceEventHandler::HandleMessage( tl::expected, JsonError>>(parsed); if (!res.has_value()) { - status_manager_.SetError( + status_manager_->SetError( DataSourceStatus::ErrorInfo::ErrorKind::kInvalidData, kErrorPatchInvalid); return MessageStatus::kInvalidMessage; @@ -193,8 +200,9 @@ DataSourceEventHandler::MessageStatus DataSourceEventHandler::HandleMessage( if (res->has_value()) { auto const& patch = (**res); auto const& key = patch.key; - std::visit([this, &key](auto&& arg) { handler_.Upsert(key, arg); }, - patch.data); + std::visit( + [&handler, &key](auto&& arg) { handler->Upsert(key, arg); }, + patch.data); return MessageStatus::kMessageHandled; } // We didn't recognize the type of the patch. So we ignore it. @@ -205,7 +213,7 @@ DataSourceEventHandler::MessageStatus DataSourceEventHandler::HandleMessage( auto parsed = boost::json::parse(data, error_code); if (error_code) { LD_LOG(logger_, LogLevel::kError) << kErrorParsingDelete; - status_manager_.SetError( + status_manager_->SetError( DataSourceStatus::ErrorInfo::ErrorKind::kInvalidData, kErrorParsingDelete); return MessageStatus::kInvalidMessage; @@ -217,13 +225,13 @@ DataSourceEventHandler::MessageStatus DataSourceEventHandler::HandleMessage( if (res.has_value()) { switch (res->kind) { case data_components::DataKind::kFlag: { - handler_.Upsert(res->key, + handler->Upsert(res->key, data_model::FlagDescriptor( data_model::Tombstone(res->version))); return MessageStatus::kMessageHandled; } case data_components::DataKind::kSegment: { - handler_.Upsert(res->key, + handler->Upsert(res->key, data_model::SegmentDescriptor( data_model::Tombstone(res->version))); return MessageStatus::kMessageHandled; @@ -233,7 +241,7 @@ DataSourceEventHandler::MessageStatus DataSourceEventHandler::HandleMessage( } } - status_manager_.SetError( + status_manager_->SetError( DataSourceStatus::ErrorInfo::ErrorKind::kInvalidData, kErrorDeleteInvalid); return MessageStatus::kInvalidMessage; diff --git a/libs/server-sdk/src/data_systems/background_sync/sources/streaming/event_handler.hpp b/libs/server-sdk/src/data_systems/background_sync/sources/streaming/event_handler.hpp index 48401ce4a..447b1bb00 100644 --- a/libs/server-sdk/src/data_systems/background_sync/sources/streaming/event_handler.hpp +++ b/libs/server-sdk/src/data_systems/background_sync/sources/streaming/event_handler.hpp @@ -11,6 +11,7 @@ #include #include +#include namespace launchdarkly::server_side::data_systems { @@ -103,9 +104,10 @@ class DataSourceEventHandler { }; DataSourceEventHandler( - data_interfaces::IDestination& handler, + std::weak_ptr handler, Logger const& logger, - data_components::DataSourceStatusManager& status_manager); + std::shared_ptr + status_manager); /** * Handles an event from the LaunchDarkly service. @@ -117,8 +119,8 @@ class DataSourceEventHandler { std::string const& data); private: - data_interfaces::IDestination& handler_; + std::weak_ptr handler_; Logger const& logger_; - data_components::DataSourceStatusManager& status_manager_; + std::shared_ptr status_manager_; }; } // namespace launchdarkly::server_side::data_systems diff --git a/libs/server-sdk/src/data_systems/background_sync/sources/streaming/streaming_data_source.cpp b/libs/server-sdk/src/data_systems/background_sync/sources/streaming/streaming_data_source.cpp index 5f7a27a7b..7d4958ea5 100644 --- a/libs/server-sdk/src/data_systems/background_sync/sources/streaming/streaming_data_source.cpp +++ b/libs/server-sdk/src/data_systems/background_sync/sources/streaming/streaming_data_source.cpp @@ -31,25 +31,25 @@ std::string const& StreamingDataSource::Identity() const { StreamingDataSource::StreamingDataSource( boost::asio::any_io_executor io, Logger const& logger, - data_components::DataSourceStatusManager& status_manager, + std::shared_ptr status_manager, config::built::ServiceEndpoints const& endpoints, config::built::BackgroundSyncConfig::StreamingConfig const& streaming, config::built::HttpProperties const& http_properties) : io_(std::move(io)), logger_(logger), - status_manager_(status_manager), + status_manager_(std::move(status_manager)), http_config_(http_properties), streaming_endpoint_(endpoints.StreamingBaseUrl()), streaming_config_(streaming) {} void StreamingDataSource::StartAsync( - data_interfaces::IDestination* dest, + std::shared_ptr dest, data_model::SDKDataSet const* bootstrap_data) { boost::ignore_unused(bootstrap_data); - event_handler_.emplace(*dest, logger_, status_manager_); + event_handler_.emplace(dest, logger_, status_manager_); - status_manager_.SetState(DataSourceStatus::DataSourceState::kInitializing); + status_manager_->SetState(DataSourceStatus::DataSourceState::kInitializing); auto updated_url = network::AppendUrl(streaming_endpoint_, streaming_config_.streaming_path); @@ -68,7 +68,7 @@ void StreamingDataSource::StartAsync( // Bad URL, don't set the client. Start will then report the bad status. if (!updated_url) { LD_LOG(logger_, LogLevel::kError) << kCouldNotParseEndpoint; - status_manager_.SetState( + status_manager_->SetState( DataSourceStatus::DataSourceState::kOff, DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError, kCouldNotParseEndpoint); @@ -80,7 +80,7 @@ void StreamingDataSource::StartAsync( // Unlikely that it could be parsed earlier, and it cannot be parsed now. if (!uri_components) { LD_LOG(logger_, LogLevel::kError) << kCouldNotParseEndpoint; - status_manager_.SetState( + status_manager_->SetState( DataSourceStatus::DataSourceState::kOff, DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError, kCouldNotParseEndpoint); @@ -129,12 +129,14 @@ void StreamingDataSource::StartAsync( if (auto self = weak_self.lock()) { auto status = self->event_handler_->HandleMessage(event.type(), event.data()); - if (status == DataSourceEventHandler::MessageStatus::kInvalidMessage) { + if (status == + DataSourceEventHandler::MessageStatus::kInvalidMessage) { // Invalid data received - restart the connection with backoff // to get a fresh stream. The backoff mechanism prevents rapid // reconnection attempts. LD_LOG(self->logger_, LogLevel::kWarn) - << "Received invalid data from stream, restarting connection"; + << "Received invalid data from stream, restarting " + "connection"; if (self->client_) { self->client_->async_restart("invalid data in stream"); } @@ -162,7 +164,7 @@ void StreamingDataSource::StartAsync( if (!client_) { LD_LOG(logger_, LogLevel::kError) << kCouldNotParseEndpoint; - status_manager_.SetState( + status_manager_->SetState( DataSourceStatus::DataSourceState::kOff, DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError, kCouldNotParseEndpoint); @@ -182,7 +184,7 @@ void StreamingDataSource::HandleErrorStateChange(sse::Error error, [this, state, error_string = std::move(error_string)](auto error) { using T = std::decay_t; if constexpr (std::is_same_v) { - this->status_manager_.SetState( + this->status_manager_->SetState( state, DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError, std::move(error_string)); @@ -190,7 +192,7 @@ void StreamingDataSource::HandleErrorStateChange(sse::Error error, } else if constexpr (std::is_same_v< T, sse::errors::UnrecoverableClientError>) { - this->status_manager_.SetState( + this->status_manager_->SetState( state, static_cast(error.status), @@ -198,14 +200,14 @@ void StreamingDataSource::HandleErrorStateChange(sse::Error error, } else if constexpr (std::is_same_v< T, sse::errors::InvalidRedirectLocation>) { - this->status_manager_.SetState( + this->status_manager_->SetState( state, DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError, std::move(error_string)); } else if constexpr (std::is_same_v) { - this->status_manager_.SetState( + this->status_manager_->SetState( state, DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError, std::move(error_string)); @@ -219,7 +221,7 @@ void StreamingDataSource::HandleErrorStateChange(sse::Error error, void StreamingDataSource::ShutdownAsync(std::function completion) { if (client_) { - status_manager_.SetState( + status_manager_->SetState( DataSourceStatus::DataSourceState::kInitializing); return client_->async_shutdown(std::move(completion)); } diff --git a/libs/server-sdk/src/data_systems/background_sync/sources/streaming/streaming_data_source.hpp b/libs/server-sdk/src/data_systems/background_sync/sources/streaming/streaming_data_source.hpp index f606686c1..4815873f4 100644 --- a/libs/server-sdk/src/data_systems/background_sync/sources/streaming/streaming_data_source.hpp +++ b/libs/server-sdk/src/data_systems/background_sync/sources/streaming/streaming_data_source.hpp @@ -12,6 +12,8 @@ #include +#include + namespace launchdarkly::server_side::data_systems { class StreamingDataSource final @@ -21,12 +23,13 @@ class StreamingDataSource final StreamingDataSource( boost::asio::any_io_executor io, Logger const& logger, - data_components::DataSourceStatusManager& status_manager, + std::shared_ptr + status_manager, config::built::ServiceEndpoints const& endpoints, config::built::BackgroundSyncConfig::StreamingConfig const& streaming, config::built::HttpProperties const& http_properties); - void StartAsync(data_interfaces::IDestination* dest, + void StartAsync(std::shared_ptr dest, data_model::SDKDataSet const* bootstrap_data) override; void ShutdownAsync(std::function completion) override; @@ -38,7 +41,7 @@ class StreamingDataSource final boost::asio::any_io_executor io_; Logger const& logger_; - data_components::DataSourceStatusManager& status_manager_; + std::shared_ptr status_manager_; config::built::HttpProperties http_config_; std::optional event_handler_; diff --git a/libs/server-sdk/src/data_systems/fdv2/fdv1_adapter_synchronizer.cpp b/libs/server-sdk/src/data_systems/fdv2/fdv1_adapter_synchronizer.cpp index 67bb98766..59932fc3f 100644 --- a/libs/server-sdk/src/data_systems/fdv2/fdv1_adapter_synchronizer.cpp +++ b/libs/server-sdk/src/data_systems/fdv2/fdv1_adapter_synchronizer.cpp @@ -123,9 +123,9 @@ std::string const& FDv1AdapterSynchronizer::ConvertingDestination::Identity() FDv1AdapterSynchronizer::FDv1AdapterSynchronizer(SourceBuilder source_builder) : state_(std::make_shared(close_promise_.GetFuture())), - destination_(std::make_unique(state_)), + destination_(std::make_shared(state_)), status_manager_( - std::make_unique()), + std::make_shared()), status_subscription_(status_manager_->OnDataSourceStatusChange( [state = state_](DataSourceStatus status) { auto error = status.LastError(); @@ -149,7 +149,7 @@ FDv1AdapterSynchronizer::FDv1AdapterSynchronizer(SourceBuilder source_builder) break; } })), - fdv1_source_(source_builder(*status_manager_)) {} + fdv1_source_(source_builder(status_manager_)) {} FDv1AdapterSynchronizer::~FDv1AdapterSynchronizer() { Close(); @@ -166,7 +166,7 @@ async::Future FDv1AdapterSynchronizer::Next( std::lock_guard lock(lifecycle_mutex_); if (!started_) { started_ = true; - fdv1_source_->StartAsync(destination_.get(), + fdv1_source_->StartAsync(destination_, /*bootstrap_data=*/nullptr); } } diff --git a/libs/server-sdk/src/data_systems/fdv2/fdv1_adapter_synchronizer.hpp b/libs/server-sdk/src/data_systems/fdv2/fdv1_adapter_synchronizer.hpp index a382ff90a..48cb77b6c 100644 --- a/libs/server-sdk/src/data_systems/fdv2/fdv1_adapter_synchronizer.hpp +++ b/libs/server-sdk/src/data_systems/fdv2/fdv1_adapter_synchronizer.hpp @@ -25,19 +25,15 @@ namespace launchdarkly::server_side::data_systems { * translated into FDv2SourceResult::ChangeSet results, with empty selectors * and fdv1_fallback = false (the directive does not re-fire from FDv1 data). * - * Threading: Next() and Close() may be called from any thread; only one - * Next() may be outstanding at a time. Member declaration order ensures - * the wrapped FDv1 source destructs before destination_ and state_, so any - * in-flight FDv1 callbacks land on live objects during teardown. This - * relies on the wrapped IDataSynchronizer blocking on its in-flight work - * in its destructor. + * Threading: Next() and Close() may be called from any thread. Only one + * Next() may be outstanding at a time. */ class FDv1AdapterSynchronizer final : public data_interfaces::IFDv2Synchronizer { public: using SourceBuilder = std::function( - data_components::DataSourceStatusManager&)>; + std::shared_ptr)>; /** * @param source_builder Called once during construction with the @@ -110,9 +106,9 @@ class FDv1AdapterSynchronizer final // shared_ptr so async callbacks that may fire after this is destroyed // can hold their own reference. std::shared_ptr const state_; - std::unique_ptr const destination_; + std::shared_ptr const destination_; - std::unique_ptr const + std::shared_ptr const status_manager_; std::unique_ptr const status_subscription_; diff --git a/libs/server-sdk/src/data_systems/fdv2/synchronizer_factories.cpp b/libs/server-sdk/src/data_systems/fdv2/synchronizer_factories.cpp index 140f64035..ed4a3f546 100644 --- a/libs/server-sdk/src/data_systems/fdv2/synchronizer_factories.cpp +++ b/libs/server-sdk/src/data_systems/fdv2/synchronizer_factories.cpp @@ -67,10 +67,11 @@ FDv1StreamingAdapterFactory::FDv1StreamingAdapterFactory( std::unique_ptr FDv1StreamingAdapterFactory::Build() { return std::make_unique( - [this](data_components::DataSourceStatusManager& status_manager) { + [this](std::shared_ptr + status_manager) { return std::make_shared( - executor_, logger_, status_manager, endpoints_, streaming_, - http_properties_); + executor_, logger_, std::move(status_manager), endpoints_, + streaming_, http_properties_); }); } @@ -89,10 +90,11 @@ FDv1PollingAdapterFactory::FDv1PollingAdapterFactory( std::unique_ptr FDv1PollingAdapterFactory::Build() { return std::make_unique( - [this](data_components::DataSourceStatusManager& status_manager) { + [this](std::shared_ptr + status_manager) { return std::make_shared( - executor_, logger_, status_manager, endpoints_, polling_, - http_properties_); + executor_, logger_, std::move(status_manager), endpoints_, + polling_, http_properties_); }); } diff --git a/libs/server-sdk/tests/data_source_event_handler_test.cpp b/libs/server-sdk/tests/data_source_event_handler_test.cpp index d2591fe8e..25859ff51 100644 --- a/libs/server-sdk/tests/data_source_event_handler_test.cpp +++ b/libs/server-sdk/tests/data_source_event_handler_test.cpp @@ -15,8 +15,8 @@ using namespace server_side::data_systems; TEST(DataSourceEventHandlerTests, HandlesEmptyPutMessage) { auto logger = logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); auto res = event_handler.HandleMessage("put", R"({"path":"/", "data":{}})"); @@ -25,14 +25,14 @@ TEST(DataSourceEventHandlerTests, HandlesEmptyPutMessage) { EXPECT_EQ(0, store->AllFlags().size()); EXPECT_EQ(0, store->AllSegments().size()); EXPECT_EQ(DataSourceStatus::DataSourceState::kValid, - manager.Status().State()); + manager->Status().State()); } TEST(DataSourceEventHandlerTests, HandlesInvalidPut) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); auto res = event_handler.HandleMessage("put", "{sorry"); @@ -41,14 +41,14 @@ TEST(DataSourceEventHandlerTests, HandlesInvalidPut) { EXPECT_EQ(0, store->AllFlags().size()); EXPECT_EQ(0, store->AllSegments().size()); EXPECT_EQ(DataSourceStatus::DataSourceState::kInitializing, - manager.Status().State()); + manager->Status().State()); } TEST(DataSourceEventHandlerTests, HandlesInvalidPatch) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); auto res = event_handler.HandleMessage("put", "{sorry"); @@ -57,42 +57,42 @@ TEST(DataSourceEventHandlerTests, HandlesInvalidPatch) { EXPECT_EQ(0, store->AllFlags().size()); EXPECT_EQ(0, store->AllSegments().size()); EXPECT_EQ(DataSourceStatus::DataSourceState::kInitializing, - manager.Status().State()); + manager->Status().State()); } TEST(DataSourceEventHandlerTests, HandlesPatchForUnknownPath) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); auto res = event_handler.HandleMessage( "patch", R"({"path":"potato", "data": "SPUD"})"); ASSERT_EQ(DataSourceEventHandler::MessageStatus::kMessageHandled, res); EXPECT_EQ(DataSourceStatus::DataSourceState::kInitializing, - manager.Status().State()); + manager->Status().State()); } TEST(DataSourceEventHandlerTests, HandlesPutForUnknownPath) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); auto res = event_handler.HandleMessage( "put", R"({"path":"potato", "data": "SPUD"})"); ASSERT_EQ(DataSourceEventHandler::MessageStatus::kMessageHandled, res); EXPECT_EQ(DataSourceStatus::DataSourceState::kInitializing, - manager.Status().State()); + manager->Status().State()); } TEST(DataSourceEventHandlerTests, HandlesInvalidDelete) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); auto res = event_handler.HandleMessage("put", "{sorry"); @@ -101,14 +101,14 @@ TEST(DataSourceEventHandlerTests, HandlesInvalidDelete) { EXPECT_EQ(0, store->AllFlags().size()); EXPECT_EQ(0, store->AllSegments().size()); EXPECT_EQ(DataSourceStatus::DataSourceState::kInitializing, - manager.Status().State()); + manager->Status().State()); } TEST(DataSourceEventHandlerTests, HandlesPayloadWithFlagAndSegment) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); auto payload = R"({"path":"/","data":{"segments":{"special":{"key":"special","included":["bob"], "version":2}},"flags":{"HasBob":{"key":"HasBob","on":true,"fallthrough": @@ -122,14 +122,14 @@ TEST(DataSourceEventHandlerTests, HandlesPayloadWithFlagAndSegment) { EXPECT_TRUE(store->GetFlag("HasBob")); EXPECT_TRUE(store->GetSegment("special")); EXPECT_EQ(DataSourceStatus::DataSourceState::kValid, - manager.Status().State()); + manager->Status().State()); } TEST(DataSourceEventHandlerTests, HandlesValidFlagPatch) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); event_handler.HandleMessage("put", "{}"); @@ -146,8 +146,8 @@ TEST(DataSourceEventHandlerTests, HandlesValidFlagPatch) { TEST(DataSourceEventHandlerTests, HandlesValidSegmentPatch) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); event_handler.HandleMessage("put", "{}"); @@ -164,8 +164,8 @@ TEST(DataSourceEventHandlerTests, HandlesValidSegmentPatch) { TEST(DataSourceEventHandlerTests, HandlesDeleteFlag) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); event_handler.HandleMessage( "put", R"({"path":"/","data":{"segments":{})" @@ -185,8 +185,8 @@ TEST(DataSourceEventHandlerTests, HandlesDeleteFlag) { TEST(DataSourceEventHandlerTests, HandlesDeleteSegment) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); event_handler.HandleMessage( "put", @@ -207,8 +207,8 @@ TEST(DataSourceEventHandlerTests, HandlesDeleteSegment) { TEST(DataSourceEventHandlerTests, HandlesPatchWithNullDataForFlag) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // Initialize the store event_handler.HandleMessage("put", R"({"path":"/", "data":{}})"); @@ -218,19 +218,20 @@ TEST(DataSourceEventHandlerTests, HandlesPatchWithNullDataForFlag) { "patch", R"({"path": "/flags/flagA", "data": null})"); ASSERT_EQ(DataSourceEventHandler::MessageStatus::kInvalidMessage, res); - // The error should be recorded, but we stay in Valid state after a previous successful PUT + // The error should be recorded, but we stay in Valid state after a previous + // successful PUT EXPECT_EQ(DataSourceStatus::DataSourceState::kValid, - manager.Status().State()); - ASSERT_TRUE(manager.Status().LastError().has_value()); + manager->Status().State()); + ASSERT_TRUE(manager->Status().LastError().has_value()); EXPECT_EQ(DataSourceStatus::ErrorInfo::ErrorKind::kInvalidData, - manager.Status().LastError()->Kind()); + manager->Status().LastError()->Kind()); } TEST(DataSourceEventHandlerTests, HandlesPatchWithNullDataForSegment) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // Initialize the store event_handler.HandleMessage("put", R"({"path":"/", "data":{}})"); @@ -240,26 +241,27 @@ TEST(DataSourceEventHandlerTests, HandlesPatchWithNullDataForSegment) { "patch", R"({"path": "/segments/segmentA", "data": null})"); ASSERT_EQ(DataSourceEventHandler::MessageStatus::kInvalidMessage, res); - // The error should be recorded, but we stay in Valid state after a previous successful PUT + // The error should be recorded, but we stay in Valid state after a previous + // successful PUT EXPECT_EQ(DataSourceStatus::DataSourceState::kValid, - manager.Status().State()); - ASSERT_TRUE(manager.Status().LastError().has_value()); + manager->Status().State()); + ASSERT_TRUE(manager->Status().LastError().has_value()); EXPECT_EQ(DataSourceStatus::ErrorInfo::ErrorKind::kInvalidData, - manager.Status().LastError()->Kind()); + manager->Status().LastError()->Kind()); } TEST(DataSourceEventHandlerTests, HandlesPatchWithMissingDataField) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // Initialize the store event_handler.HandleMessage("put", R"({"path":"/", "data":{}})"); // Missing data field should also be treated as invalid - auto res = event_handler.HandleMessage( - "patch", R"({"path": "/flags/flagA"})"); + auto res = + event_handler.HandleMessage("patch", R"({"path": "/flags/flagA"})"); ASSERT_EQ(DataSourceEventHandler::MessageStatus::kInvalidMessage, res); } @@ -267,12 +269,12 @@ TEST(DataSourceEventHandlerTests, HandlesPatchWithMissingDataField) { TEST(DataSourceEventHandlerTests, HandlesPutWithNullData) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // PUT with null data should also be handled safely - auto res = event_handler.HandleMessage( - "put", R"({"path":"/", "data": null})"); + auto res = + event_handler.HandleMessage("put", R"({"path":"/", "data": null})"); // PUT handles this differently - it may succeed with empty data // The important thing is it doesn't crash @@ -285,8 +287,8 @@ TEST(DataSourceEventHandlerTests, HandlesPutWithNullData) { TEST(DataSourceEventHandlerTests, HandlesPatchWithBooleanData) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // Initialize the store event_handler.HandleMessage("put", R"({"path":"/", "data":{}})"); @@ -301,8 +303,8 @@ TEST(DataSourceEventHandlerTests, HandlesPatchWithBooleanData) { TEST(DataSourceEventHandlerTests, HandlesPatchWithStringData) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // Initialize the store event_handler.HandleMessage("put", R"({"path":"/", "data":{}})"); @@ -317,8 +319,8 @@ TEST(DataSourceEventHandlerTests, HandlesPatchWithStringData) { TEST(DataSourceEventHandlerTests, HandlesPatchWithArrayData) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // Initialize the store event_handler.HandleMessage("put", R"({"path":"/", "data":{}})"); @@ -333,8 +335,8 @@ TEST(DataSourceEventHandlerTests, HandlesPatchWithArrayData) { TEST(DataSourceEventHandlerTests, HandlesPatchWithNumberData) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // Initialize the store event_handler.HandleMessage("put", R"({"path":"/", "data":{}})"); @@ -349,8 +351,8 @@ TEST(DataSourceEventHandlerTests, HandlesPatchWithNumberData) { TEST(DataSourceEventHandlerTests, HandlesDeleteWithStringVersion) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // Initialize the store event_handler.HandleMessage("put", R"({"path":"/", "data":{}})"); @@ -365,8 +367,8 @@ TEST(DataSourceEventHandlerTests, HandlesDeleteWithStringVersion) { TEST(DataSourceEventHandlerTests, HandlesPutWithInvalidFlagsType) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // Flags should be an object, not a boolean auto res = event_handler.HandleMessage( @@ -378,8 +380,8 @@ TEST(DataSourceEventHandlerTests, HandlesPutWithInvalidFlagsType) { TEST(DataSourceEventHandlerTests, HandlesPutWithInvalidSegmentsType) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // Segments should be an object, not an array auto res = event_handler.HandleMessage( @@ -393,8 +395,8 @@ TEST(DataSourceEventHandlerTests, HandlesPutWithInvalidSegmentsType) { TEST(DataSourceEventHandlerTests, HandlesUnterminatedString) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // Unterminated string should be treated as malformed JSON auto res = event_handler.HandleMessage( @@ -406,8 +408,8 @@ TEST(DataSourceEventHandlerTests, HandlesUnterminatedString) { TEST(DataSourceEventHandlerTests, HandlesTrailingComma) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // Trailing comma should be treated as malformed JSON auto res = event_handler.HandleMessage( @@ -421,15 +423,14 @@ TEST(DataSourceEventHandlerTests, HandlesTrailingComma) { TEST(DataSourceEventHandlerTests, HandlesDeleteWithMissingPath) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // Initialize the store event_handler.HandleMessage("put", R"({"path":"/", "data":{}})"); // Missing path field should be treated as invalid - auto res = event_handler.HandleMessage( - "delete", R"({"version": 1})"); + auto res = event_handler.HandleMessage("delete", R"({"version": 1})"); ASSERT_EQ(DataSourceEventHandler::MessageStatus::kInvalidMessage, res); } @@ -437,15 +438,15 @@ TEST(DataSourceEventHandlerTests, HandlesDeleteWithMissingPath) { TEST(DataSourceEventHandlerTests, HandlesDeleteWithMissingVersion) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // Initialize the store event_handler.HandleMessage("put", R"({"path":"/", "data":{}})"); // Missing version field should be treated as invalid - auto res = event_handler.HandleMessage( - "delete", R"({"path": "/flags/flagA"})"); + auto res = + event_handler.HandleMessage("delete", R"({"path": "/flags/flagA"})"); ASSERT_EQ(DataSourceEventHandler::MessageStatus::kInvalidMessage, res); } @@ -453,13 +454,12 @@ TEST(DataSourceEventHandlerTests, HandlesDeleteWithMissingVersion) { TEST(DataSourceEventHandlerTests, HandlesPutWithMissingPath) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); // Missing/empty path is treated as unrecognized (safely ignored) // This provides forward compatibility - auto res = event_handler.HandleMessage( - "put", R"({"data": {}})"); + auto res = event_handler.HandleMessage("put", R"({"data": {}})"); ASSERT_EQ(DataSourceEventHandler::MessageStatus::kMessageHandled, res); } @@ -467,11 +467,11 @@ TEST(DataSourceEventHandlerTests, HandlesPutWithMissingPath) { TEST(DataSourceEventHandlerTests, HandlesEmptyJsonObject) { auto logger = launchdarkly::logging::NullLogger(); auto store = std::make_shared(); - DataSourceStatusManager manager; - DataSourceEventHandler event_handler(*store, logger, manager); + auto manager = std::make_shared(); + DataSourceEventHandler event_handler(store, logger, manager); - // Empty JSON object with missing path is treated as unrecognized (safely ignored) - // This provides forward compatibility with future event types + // Empty JSON object with missing path is treated as unrecognized (safely + // ignored) This provides forward compatibility with future event types auto res = event_handler.HandleMessage("patch", "{}"); ASSERT_EQ(DataSourceEventHandler::MessageStatus::kMessageHandled, res); diff --git a/libs/server-sdk/tests/fdv1_adapter_synchronizer_test.cpp b/libs/server-sdk/tests/fdv1_adapter_synchronizer_test.cpp index 537afe0c7..c42e2d452 100644 --- a/libs/server-sdk/tests/fdv1_adapter_synchronizer_test.cpp +++ b/libs/server-sdk/tests/fdv1_adapter_synchronizer_test.cpp @@ -24,12 +24,13 @@ namespace { // the IDestination it was given so the test can drive Init/Upsert. class MockFDv1Source final : public IDataSynchronizer { public: - explicit MockFDv1Source(DataSourceStatusManager& /*status_manager*/) {} + explicit MockFDv1Source( + std::shared_ptr /*status_manager*/) {} - void StartAsync(IDestination* destination, + void StartAsync(std::shared_ptr destination, data_model::SDKDataSet const* bootstrap) override { ++start_count; - destination_ = destination; + destination_ = std::move(destination); bootstrap_was_null = (bootstrap == nullptr); } @@ -45,7 +46,7 @@ class MockFDv1Source final : public IDataSynchronizer { return id; } - IDestination* destination_ = nullptr; + std::shared_ptr destination_; int start_count = 0; int shutdown_count = 0; bool bootstrap_was_null = false; @@ -57,9 +58,9 @@ class MockFDv1Source final : public IDataSynchronizer { FDv1AdapterSynchronizer::SourceBuilder MakeMockBuilder( MockFDv1Source** out_source = nullptr, DataSourceStatusManager** out_sm = nullptr) { - return [out_source, out_sm](DataSourceStatusManager& sm) { + return [out_source, out_sm](std::shared_ptr sm) { if (out_sm) { - *out_sm = &sm; + *out_sm = sm.get(); } auto source = std::make_shared(sm); if (out_source) { @@ -81,9 +82,9 @@ class DeferredCompletionFDv1Source final explicit DeferredCompletionFDv1Source(boost::asio::any_io_executor executor) : executor_(std::move(executor)) {} - void StartAsync(IDestination* destination, + void StartAsync(std::shared_ptr destination, data_model::SDKDataSet const* /*bootstrap*/) override { - destination_ = destination; + destination_ = std::move(destination); } // Posts the drain: one in-flight upsert into the destination, then the @@ -110,7 +111,7 @@ class DeferredCompletionFDv1Source final } boost::asio::any_io_executor executor_; - IDestination* destination_ = nullptr; + std::shared_ptr destination_; bool completion_invoked = false; }; @@ -127,7 +128,9 @@ TEST(FDv1AdapterSynchronizerTest, DestinationStaysValidUntilShutdownCompletes) { std::make_shared(ioc.get_executor()); { FDv1AdapterSynchronizer adapter( - [source](DataSourceStatusManager&) { return source; }); + [source](std::shared_ptr) { + return source; + }); adapter.Next(data_model::Selector{}); // triggers StartAsync } // adapter destroyed: Close() requests shutdown, the drain is still // pending