fix: Harden FDv1 data source destination and status-manager lifetimes

libs/server-sdk/src/client_impl.cpp

@@ -70,7 +70,7 @@
     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<data_components::DataSourceStatusManager> status_manager,
     Logger& logger) {
     return std::make_unique<data_systems::BackgroundSync>(
         endpoints, cfg, http_properties, executor, status_manager, logger);
@@ -94,7 +94,7 @@
     std::vector<std::unique_ptr<data_interfaces::IFDv2InitializerFactory>>
         initializer_factories;
     for (auto const& initializer : cfg.initializers) {
         initializer_factories.push_back(
             std::make_unique<data_systems::FDv2PollingInitializerFactory>(
                 executor, logger, endpoints, http_properties, initializer));
     }
@@ -161,10 +161,10 @@
     config::built::HttpProperties const& http_properties,
     Config const& config,
     boost::asio::any_io_executor const& executor,
-    data_components::DataSourceStatusManager& status_manager,
+    std::shared_ptr<data_components::DataSourceStatusManager> status_manager,
     Logger& logger) {
     if (config.DataSystemConfig().disabled) {
-        return std::make_unique<data_systems::OfflineSystem>(status_manager);
+        return std::make_unique<data_systems::OfflineSystem>(*status_manager);
     }
 
     auto data_source_properties =
@@ -178,12 +178,12 @@
                     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_);
@@ -220,7 +220,7 @@
 bool IsFlagPresent(
     std::shared_ptr<data_model::FlagDescriptor> const& flag_desc);
 
 ClientImpl::ClientImpl(Config config, std::string const& version)
     : config_(config),
       http_properties_(
           config::builders::HttpPropertiesBuilder(config.HttpProperties())
@@ -238,7 +238,8 @@
       logger_(MakeLogger(config.Logging())),
       ioc_(kAsioConcurrencyHint),
       work_(boost::asio::make_work_guard(ioc_)),
-      status_manager_(),
+      status_manager_(
+          std::make_shared<data_components::DataSourceStatusManager>()),
       data_system_(MakeDataSystem(http_properties_,
                                   config_,
                                   ioc_.get_executor(),
@@ -251,7 +252,7 @@
       big_segment_store_(
           config_.BigSegments()
               ? std::make_shared<data_components::BigSegmentStoreWrapper>(
                     *config_.BigSegments(),
                     ioc_.get_executor(),
                     logger_)
               : nullptr),
@@ -287,10 +288,10 @@
 }
 
 std::future<bool> ClientImpl::StartAsync() {
     auto pr = std::make_shared<std::promise<bool>>();
     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
@@ -328,14 +329,14 @@
     // system to fetch them all at once up-front. This may be a no-op
     // depending on the data system (e.g. if the segments are all already in
     // memory.)
     auto _ = data_system_->AllSegments();
 
     for (auto const& [key, v] : all_flags) {
         if (!v || !v->item) {
             continue;
         }
 
         auto const& flag = *(v->item);
 
         if (IsSet(options, AllFlagsState::Options::ClientSideOnly) &&
             !flag.clientSideAvailability.usingEnvironmentId) {
@@ -531,9 +532,9 @@
     }
 
     EvaluationDetail<Value> result =
         evaluator_.Evaluate(*flag_rule->item, context, event_scope);
     auto detail = PostEvaluation(key, context, default_value, result,
                                  event_scope, flag_rule.get()->item);
 
     // Execute afterEvaluation hooks
     if (executor) {
@@ -778,7 +779,7 @@
 }
 
 IDataSourceStatusProvider& ClientImpl::DataSourceStatus() {
-    return status_manager_;
+    return *status_manager_;
 }
 
 IBigSegmentStoreStatusProvider& ClientImpl::BigSegmentStoreStatus() {

libs/server-sdk/src/client_impl.hpp

@@ -250,7 +250,7 @@ class ClientImpl : public IClient {
     boost::asio::executor_work_guard<boost::asio::io_context::executor_type>
         work_;
 
-    data_components::DataSourceStatusManager status_manager_;
+    std::shared_ptr<data_components::DataSourceStatusManager> status_manager_;
 
     // This is the main polymorphic component that constitutes the
     // guts of how data is retrieved (polling, streaming, persistent stores,

libs/server-sdk/src/data_interfaces/source/idata_synchronizer.hpp

@@ -3,6 +3,7 @@
 #include <launchdarkly/data_model/sdk_data_set.hpp>
 
 #include <functional>
+#include <memory>
 #include <optional>
 #include <string>
 
@@ -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<IDestination> destination,
                             data_model::SDKDataSet const* bootstrap_data) = 0;
 
     /**

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<data_components::DataSourceStatusManager> status_manager,
     Logger const& logger)
-    : store_(), change_notifier_(store_, store_), synchronizer_() {
+    : store_(),
+      change_notifier_(
+          std::make_shared<data_components::ChangeNotifier>(store_, store_)),
+      synchronizer_() {
     std::visit(
         [&](auto&& method_config) {
             using T = std::decay_t<decltype(method_config)>;
@@ -34,7 +37,7 @@ BackgroundSync::BackgroundSync(
 }
 
 void BackgroundSync::Initialize() {
-    synchronizer_->StartAsync(&change_notifier_,
+    synchronizer_->StartAsync(change_notifier_,
                               nullptr /* no bootstrap data supported yet */);
 }
 

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<data_components::DataSourceStatusManager>
+            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<data_components::ChangeNotifier> change_notifier_;
     // Needs to be shared to that the source can keep itself alive through
     // async operations.
     std::shared_ptr<data_interfaces::IDataSynchronizer> synchronizer_;

libs/server-sdk/src/data_systems/background_sync/sources/polling/polling_data_source.cpp

@@ -3,8 +3,8 @@
 #include <launchdarkly/encoding/base_64.hpp>
 #include <launchdarkly/network/http_error_messages.hpp>
 
-#include <launchdarkly/serialization/json_flag.hpp>
 #include <launchdarkly/detail/serialization/json_primitives.hpp>
+#include <launchdarkly/serialization/json_flag.hpp>
 #include <launchdarkly/serialization/json_sdk_data_set.hpp>
 #include <launchdarkly/server_side/data_source_status.hpp>
 
@@ -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<data_components::DataSourceStatusManager> 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<data_model::SDKDataSet, JsonError>>(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<data_interfaces::IDestination> 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<void()> completion) {
-    status_manager_.SetState(DataSourceStatus::DataSourceState::kInitializing);
+    status_manager_->SetState(DataSourceStatus::DataSourceState::kInitializing);
     timer_.cancel();
     if (completion) {
         boost::asio::post(timer_.get_executor(), completion);

libs/server-sdk/src/data_systems/background_sync/sources/polling/polling_data_source.hpp

@@ -12,6 +12,7 @@
 #include <boost/asio/any_io_executor.hpp>
 
 #include <chrono>
+#include <memory>
 
 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<data_components::DataSourceStatusManager>
+                          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<data_interfaces::IDestination> dest,
                     data_model::SDKDataSet const* bootstrap_data) override;
 
     void ShutdownAsync(std::function<void()> 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<data_components::DataSourceStatusManager> status_manager_;
 
     // Responsible for performing HTTP requests using boost::asio.
     network::AsioRequester requester_;
@@ -66,7 +65,7 @@ class PollingDataSource
     std::chrono::time_point<std::chrono::system_clock> last_poll_start_;
 
     // Destination for all data obtained via polling.
-    data_interfaces::IDestination* sink_;
+    std::weak_ptr<data_interfaces::IDestination> sink_;
 
     void StartPollingTimer();
 };