diff --git a/libs/client-sdk/src/CMakeLists.txt b/libs/client-sdk/src/CMakeLists.txt index dcf8025a4..1ff2452b4 100644 --- a/libs/client-sdk/src/CMakeLists.txt +++ b/libs/client-sdk/src/CMakeLists.txt @@ -21,6 +21,7 @@ target_sources(${LIBNAME} PRIVATE data_sources/fdv2/polling_initializer.cpp data_sources/fdv2/polling_synchronizer.cpp data_sources/fdv2/streaming_synchronizer.cpp + data_sources/fdv2/fdv2_data_source.cpp data_sources/data_source_event_handler.cpp data_sources/polling_data_source.cpp flag_manager/flag_store.cpp @@ -47,6 +48,7 @@ target_sources(${LIBNAME} PRIVATE data_sources/fdv2/polling_initializer.hpp data_sources/fdv2/polling_synchronizer.hpp data_sources/fdv2/streaming_synchronizer.hpp + data_sources/fdv2/fdv2_data_source.hpp flag_manager/flag_store.hpp flag_manager/flag_updater.hpp bindings/c/sdk.cpp diff --git a/libs/client-sdk/src/data_sources/fdv2/fdv2_data_source.cpp b/libs/client-sdk/src/data_sources/fdv2/fdv2_data_source.cpp new file mode 100644 index 000000000..6537d6753 --- /dev/null +++ b/libs/client-sdk/src/data_sources/fdv2/fdv2_data_source.cpp @@ -0,0 +1,504 @@ +#include "fdv2_data_source.hpp" + +#include + +#include + +#include +#include +#include +#include + +namespace launchdarkly::client_side::data_sources { + +namespace { + +// Lets std::visit dispatch to a different lambda per variant alternative. +template +struct overloaded : Ts... { + using Ts::operator()...; +}; +template +overloaded(Ts...) -> overloaded; + +// Reduces a source result to the signal the conditions act on. +SourceSignal ClassifyResult(FDv2SourceResult const& result) { + if (std::get_if(&result.value)) { + return SourceSignal::kChangeSet; + } + if (std::get_if(&result.value)) { + return SourceSignal::kInterrupted; + } + return SourceSignal::kOther; +} + +bool AllFromCache( + std::vector> const& factories) { + return std::all_of( + factories.begin(), factories.end(), + [](auto const& factory) { return factory->IsFromCache(); }); +} + +} // namespace + +FDv2DataSource::FDv2DataSource( + std::vector> initializer_factories, + std::vector> + synchronizer_factories, + std::unique_ptr fallback_condition_factory, + std::unique_ptr recovery_condition_factory, + boost::asio::any_io_executor executor, + Context context, + IDataSourceUpdateSink* sink, + flag_manager::FlagStore const* store, + DataSourceStatusManager* status_manager, + Logger const& logger) + : logger_(logger), + executor_(std::move(executor)), + initializer_factories_(std::move(initializer_factories)), + fallback_condition_factory_(std::move(fallback_condition_factory)), + recovery_condition_factory_(std::move(recovery_condition_factory)), + context_(std::move(context)), + cache_only_(!initializer_factories_.empty() && + synchronizer_factories.empty() && + AllFromCache(initializer_factories_)), + sink_(sink), + store_(store), + status_manager_(status_manager), + start_called_(false), + last_logged_synchronizer_interrupted_(false), + closed_(false), + received_data_(false), + initializer_index_(0), + active_initializer_from_cache_(false), + source_manager_(std::move(synchronizer_factories)), + active_initializer_(nullptr), + active_synchronizer_(nullptr), + active_conditions_(nullptr) {} + +FDv2DataSource::~FDv2DataSource() { + Close(); +} + +void FDv2DataSource::Close() { + std::lock_guard lock(mutex_); + closed_ = true; + if (active_initializer_) { + active_initializer_->Close(); + } + if (active_synchronizer_) { + active_synchronizer_->Close(); + } + if (active_conditions_) { + active_conditions_->Close(); + } +} + +std::optional FDv2DataSource::EnvironmentId() const { + std::lock_guard lock(mutex_); + return environment_id_; +} + +void FDv2DataSource::PublishState(DataSourceStatus::DataSourceState state) { + { + std::lock_guard lock(mutex_); + if (closed_) { + return; + } + } + status_manager_->SetState(state); +} + +void FDv2DataSource::PublishState(DataSourceStatus::DataSourceState state, + DataSourceStatus::ErrorInfo::ErrorKind kind, + std::string message) { + { + std::lock_guard lock(mutex_); + if (closed_) { + return; + } + } + status_manager_->SetState(state, kind, std::move(message)); +} + +void FDv2DataSource::Start() { + bool const already_called = start_called_.exchange(true); + assert(!already_called && "Start() must be called at most once"); + + PublishState(DataSourceStatus::DataSourceState::kInitializing); + + LD_LOG(logger_, LogLevel::kInfo) << "fdv2: starting"; + if (initializer_factories_.empty() && + source_manager_.SynchronizerCount() == 0) { + // Nothing is configured to supply data, so an empty store is the + // canonical state. + PublishState(DataSourceStatus::DataSourceState::kValid); + return; + } + + // Evaluation can use cached flags as soon as Start() returns, the way it + // could when the client loaded the cache in its constructor. + RunCacheInitializers(); + + boost::asio::post(executor_, [weak = weak_from_this()]() { + if (auto self = weak.lock()) { + self->RunNextInitializer(); + } + }); +} + +void FDv2DataSource::RunCacheInitializers() { + while (true) { + std::unique_ptr initializer; + { + std::lock_guard lock(mutex_); + if (closed_ || + initializer_index_ >= initializer_factories_.size() || + !initializer_factories_[initializer_index_]->IsFromCache()) { + return; + } + initializer = initializer_factories_[initializer_index_]->Build(); + } + + auto future = initializer->Run(); + if (!future.IsFinished()) { + // Nothing here can wait on it, so the chain runs this initializer + // instead. The index is left where it is. + initializer->Close(); + return; + } + + { + std::lock_guard lock(mutex_); + ++initializer_index_; + } + + auto result = future.GetResult(); + if (result) { + if (auto* change_set = + std::get_if(&result->value)) { + LD_LOG(logger_, LogLevel::kInfo) + << "fdv2: applying cached data from " + << initializer->Identity(); + ApplyResult(std::move(*change_set), + std::move(result->environment_id), + /* from_cache= */ true); + } + } + initializer->Close(); + } +} + +void FDv2DataSource::ShutdownAsync(std::function completion) { + // Report initializing so that a caller waiting on the next status change, + // such as identify, sees the restart. This runs before Close(), which + // stops any further transitions from this source. + PublishState(DataSourceStatus::DataSourceState::kInitializing); + Close(); + if (completion) { + boost::asio::post(executor_, std::move(completion)); + } +} + +void FDv2DataSource::RunNextInitializer() { + bool exhausted = false; + { + std::lock_guard lock(mutex_); + if (closed_) { + return; + } + if (initializer_index_ >= initializer_factories_.size()) { + exhausted = true; + } else { + auto& factory = initializer_factories_[initializer_index_++]; + active_initializer_from_cache_ = factory->IsFromCache(); + active_initializer_ = factory->Build(); + LD_LOG(logger_, LogLevel::kInfo) << "fdv2: starting initializer " + << active_initializer_->Identity(); + active_initializer_->Run().Then( + [weak = weak_from_this()]( + FDv2SourceResult const& result) -> std::monostate { + if (auto self = weak.lock()) { + self->OnInitializerResult(result); + } + return {}; + }, + [executor = executor_](async::Continuation work) { + boost::asio::post(executor, std::move(work)); + }); + } + } + + if (exhausted) { + StartSynchronizers(); + } +} + +void FDv2DataSource::OnInitializerResult(FDv2SourceResult result) { + bool got_basis = false; + bool got_shutdown = false; + bool from_cache = false; + { + std::lock_guard lock(mutex_); + from_cache = active_initializer_from_cache_; + } + + std::visit( + overloaded{ + [&](FDv2SourceResult::ChangeSet& cs) { + bool const has_selector = + cs.change_set.selector.value.has_value(); + ApplyResult(std::move(cs), std::move(result.environment_id), + from_cache); + if (has_selector) { + LD_LOG(logger_, LogLevel::kInfo) + << "fdv2: initializer succeeded"; + got_basis = true; + } + }, + [&](FDv2SourceResult::Shutdown&) { got_shutdown = true; }, + [&](FDv2SourceResult::Interrupted const& iv) { + LD_LOG(logger_, LogLevel::kWarn) + << "fdv2: initializer interrupted: " << iv.error.Message(); + PublishState(DataSourceStatus::DataSourceState::kInterrupted, + iv.error.Kind(), iv.error.Message()); + }, + [&](FDv2SourceResult::TerminalError const& te) { + LD_LOG(logger_, LogLevel::kWarn) + << "fdv2: initializer terminal error: " + << te.error.Message(); + PublishState(DataSourceStatus::DataSourceState::kInterrupted, + te.error.Kind(), te.error.Message()); + }, + [&](FDv2SourceResult::Goodbye const&) { + LD_LOG(logger_, LogLevel::kDebug) + << "fdv2: ignoring goodbye from initializer"; + }, + }, + result.value); + + { + std::lock_guard lock(mutex_); + active_initializer_.reset(); + if (closed_ || got_shutdown) { + return; + } + } + + if (got_basis) { + StartSynchronizers(); + } else { + RunNextInitializer(); + } +} + +void FDv2DataSource::StartSynchronizers() { + bool exhausted = false; + bool any_synchronizers_configured = false; + { + std::lock_guard lock(mutex_); + if (closed_) { + return; + } + active_synchronizer_ = source_manager_.NextSynchronizer(); + if (active_synchronizer_) { + LD_LOG(logger_, LogLevel::kInfo) + << "fdv2: starting synchronizer " + << active_synchronizer_->Identity(); + last_logged_synchronizer_interrupted_.store(false); + active_conditions_ = BuildActiveConditions(); + } else { + exhausted = true; + any_synchronizers_configured = + source_manager_.SynchronizerCount() > 0; + } + } + + if (exhausted) { + ReportExhausted(any_synchronizers_configured); + return; + } + + RunSynchronizerNext(); +} + +void FDv2DataSource::ReportExhausted(bool any_synchronizers_configured) { + if (cache_only_) { + // The cache is the only thing that could ever have supplied data, so + // a miss is not a failure to initialize. It just means there are no + // flags. + PublishState(DataSourceStatus::DataSourceState::kValid); + return; + } + + bool received_data = false; + { + std::lock_guard lock(mutex_); + received_data = received_data_; + } + if (!any_synchronizers_configured && received_data) { + // The initializers supplied data and nothing is configured to keep it + // current, which is a complete, successful run. + return; + } + + std::string const message = + any_synchronizers_configured + ? "all data source acquisition methods have been exhausted" + : "all initializers exhausted and no synchronizers configured"; + LD_LOG(logger_, LogLevel::kWarn) << "fdv2: " << message; + PublishState(DataSourceStatus::DataSourceState::kShutdown, + DataSourceStatus::ErrorInfo::ErrorKind::kUnknown, message); +} + +void FDv2DataSource::RunSynchronizerNext() { + std::lock_guard lock(mutex_); + if (closed_ || !active_synchronizer_) { + return; + } + auto next_future = active_synchronizer_->Next(store_->CurrentSelector()); + auto cond_cancel = std::make_shared(); + auto cond_future = active_conditions_->GetFuture(cond_cancel->GetToken()); + async::WhenAny(cond_future, next_future) + .Then( + [weak = weak_from_this(), cond_future, next_future, + cond_cancel](std::size_t const& idx) -> std::monostate { + cond_cancel->Cancel(); + auto self = weak.lock(); + if (!self) { + return {}; + } + if (idx == 0) { + self->OnConditionFired(*cond_future.GetResult()); + } else { + self->OnSynchronizerResult(*next_future.GetResult()); + } + return {}; + }, + [executor = executor_](async::Continuation work) { + boost::asio::post(executor, std::move(work)); + }); +} + +void FDv2DataSource::OnConditionFired(IFDv2Condition::Type type) { + if (type == IFDv2Condition::Type::kCancelled) { + return; + } + { + std::lock_guard lock(mutex_); + if (closed_) { + return; + } + // Destructors close the active synchronizer and conditions. + active_synchronizer_.reset(); + active_conditions_.reset(); + if (type == IFDv2Condition::Type::kRecovery) { + LD_LOG(logger_, LogLevel::kInfo) << "fdv2: recovery condition met"; + source_manager_.ResetSourceIndex(); + } else { + LD_LOG(logger_, LogLevel::kInfo) << "fdv2: fallback condition met"; + } + } + StartSynchronizers(); +} + +std::unique_ptr FDv2DataSource::BuildActiveConditions() const { + std::vector> conditions; + // With only one synchronizer available there's nothing to fall back to + // or recover from, so leave the conditions empty. + if (source_manager_.AvailableSynchronizerCount() == 1) { + return std::make_unique(std::move(conditions)); + } + if (fallback_condition_factory_) { + conditions.push_back(fallback_condition_factory_->Build()); + } + // The prime synchronizer has nothing more-preferred to recover to. + if (!source_manager_.IsPrimeSynchronizer() && recovery_condition_factory_) { + conditions.push_back(recovery_condition_factory_->Build()); + } + return std::make_unique(std::move(conditions)); +} + +void FDv2DataSource::OnSynchronizerResult(FDv2SourceResult result) { + { + std::lock_guard lock(mutex_); + if (closed_) { + return; + } + if (active_conditions_) { + active_conditions_->Inform(ClassifyResult(result)); + } + } + + bool got_shutdown = false; + bool advance = false; + + std::visit( + overloaded{ + [&](FDv2SourceResult::ChangeSet& cs) { + last_logged_synchronizer_interrupted_.store(false); + ApplyResult(std::move(cs), std::move(result.environment_id), + /* from_cache= */ false); + }, + [&](FDv2SourceResult::Shutdown&) { got_shutdown = true; }, + [&](FDv2SourceResult::Interrupted const& iv) { + if (!last_logged_synchronizer_interrupted_.exchange(true)) { + LD_LOG(logger_, LogLevel::kInfo) + << "fdv2: synchronizer interrupted: " + << iv.error.Message(); + } + PublishState(DataSourceStatus::DataSourceState::kInterrupted, + iv.error.Kind(), iv.error.Message()); + }, + [&](FDv2SourceResult::TerminalError const& te) { + LD_LOG(logger_, LogLevel::kWarn) + << "fdv2: synchronizer terminal error: " + << te.error.Message(); + PublishState(DataSourceStatus::DataSourceState::kInterrupted, + te.error.Kind(), te.error.Message()); + advance = true; + }, + [&](FDv2SourceResult::Goodbye const&) { + // The synchronizer restarts its own connection. + }, + }, + result.value); + + { + std::lock_guard lock(mutex_); + if (closed_ || got_shutdown) { + active_synchronizer_.reset(); + active_conditions_.reset(); + return; + } + if (advance) { + source_manager_.BlockCurrentSynchronizer(); + active_synchronizer_.reset(); + active_conditions_.reset(); + } + } + + if (advance) { + StartSynchronizers(); + } else { + RunSynchronizerNext(); + } +} + +void FDv2DataSource::ApplyResult(FDv2SourceResult::ChangeSet change_set, + std::optional environment_id, + bool from_cache) { + bool const carries_data = + change_set.change_set.type != data_model::ChangeSetType::kNone; + { + std::lock_guard lock(mutex_); + if (environment_id) { + environment_id_ = std::move(environment_id); + } + received_data_ = received_data_ || carries_data; + } + sink_->Apply(context_, std::move(change_set.change_set), from_cache); + PublishState(DataSourceStatus::DataSourceState::kValid); +} + +} // namespace launchdarkly::client_side::data_sources diff --git a/libs/client-sdk/src/data_sources/fdv2/fdv2_data_source.hpp b/libs/client-sdk/src/data_sources/fdv2/fdv2_data_source.hpp new file mode 100644 index 000000000..198c3024d --- /dev/null +++ b/libs/client-sdk/src/data_sources/fdv2/fdv2_data_source.hpp @@ -0,0 +1,229 @@ +#pragma once + +#include "../data_source.hpp" +#include "../data_source_status_manager.hpp" +#include "../data_source_update_sink.hpp" +#include "ifdv2_initializer_factory.hpp" +#include "ifdv2_synchronizer_factory.hpp" + +#include "../../flag_manager/flag_store.hpp" + +#include +#include +#include +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include + +namespace launchdarkly::client_side::data_sources { + +// The orchestration primitives the client and server SDKs share. +using internal::data_sources::Conditions; +using internal::data_sources::FallbackConditionFactory; +using internal::data_sources::IFDv2Condition; +using internal::data_sources::IFDv2ConditionFactory; +using internal::data_sources::RecoveryConditionFactory; +using internal::data_sources::SourceSignal; + +using SourceManager = + internal::data_sources::SourceManager; + +/** + * The FDv2 data source. It runs a sequence of initializers to load flag data + * for one evaluation context, then hands off to a synchronizer to keep that + * data current, rotating synchronizers as they fail and recover. + * + * The data source is built for a single evaluation context. Changing context + * means shutting this one down and starting another. + * + * Lifecycle: + * 1. Construct. + * 2. Call Start() exactly once. It returns immediately, and orchestration + * runs on the executor. + * 3. Call ShutdownAsync() to stop. The completion runs on the executor once + * the source has stopped touching the store. + * + * Thread safety: Start, ShutdownAsync, and EnvironmentId may be called from + * any thread. + * + * Orchestration: + * + * Start() + * | + * v + * +-------------------+ no sources configured + * | Anything to do? |---------> [Done, status = kValid] + * +-------------------+ + * | + * v + * +-------------------+ initializer #N returns: + * | Initializer phase| ChangeSet(no selector) -> stay, N += 1 + * | N = 0, 1, 2, ... | ChangeSet(selector) -> go to Sync + * | | Interrupted/Terminal -> stay, N += 1 + * | | Goodbye -> stay, N += 1 + * | | Shutdown -> [Closed] + * +-------------------+ + * | + * | (N exhausted, or basis received) + * v + * +-------------------+ active synchronizer's Next returns: + * | Synchronizer | ChangeSet -> apply, loop + * | phase | Interrupted -> loop (source self-retries) + * | (cyclic; | Goodbye -> loop (source self-restarts) + * | blocked sources | TerminalError -> block, advance + * | are skipped) | Shutdown -> [Closed] + * +-------------------+ + * ^ | fallback condition -> advance (with wrap) + * | | recovery condition -> reset to first available + * +---+ + * | + * | (all synchronizers blocked) + * v + * [Done; final status preserved] + */ +class FDv2DataSource final + : public IDataSource, + public std::enable_shared_from_this { + public: + /** + * @param initializer_factories Build the initializers, run in order to + * load a basis. + * @param synchronizer_factories Build the synchronizers, used in order to + * keep data current once initialization is done. + * @param fallback_condition_factory Builds the per-synchronizer fallback + * condition. May be null, in which case synchronizers rotate only on + * terminal errors. + * @param recovery_condition_factory Builds the per-synchronizer recovery + * condition. May be null, in which case the source never returns to a + * more-preferred synchronizer once it has fallen back. + * @param executor Runs the orchestration. + * @param context The evaluation context this source loads data for. + * @param sink Receives the changesets. Non-owning. Must outlive this + * object. + * @param store Supplies the selector to request incremental updates + * against. Non-owning. Must outlive this object. + * @param status_manager Publishes data source status transitions. + * Non-owning. Must outlive this object. + * @param logger Receives diagnostic logging. + */ + FDv2DataSource( + std::vector> + initializer_factories, + std::vector> + synchronizer_factories, + std::unique_ptr fallback_condition_factory, + std::unique_ptr recovery_condition_factory, + boost::asio::any_io_executor executor, + Context context, + IDataSourceUpdateSink* sink, + flag_manager::FlagStore const* store, + DataSourceStatusManager* status_manager, + Logger const& logger); + + ~FDv2DataSource() override; + + void Start() override; + + void ShutdownAsync(std::function completion) override; + + /** + * The environment the service reported the most recent payload was + * evaluated in, or nullopt if no response has reported one. + */ + [[nodiscard]] std::optional EnvironmentId() const; + + private: + /** + * Signals the orchestration to stop and closes any active source. + * Idempotent. + */ + void Close(); + + // Orchestration steps. Each chains the next through Future::Then, so at + // most one step has a pending continuation at any time. mutex_ provides + // mutual exclusion for orchestration state, and lets Close() tear down + // active sources from any thread. + + // Publishes a status transition unless Close() has run. The client drops + // a data source as soon as its replacement starts, and a dropped source + // must not report over the new one. + void PublishState(DataSourceStatus::DataSourceState state); + void PublishState(DataSourceStatus::DataSourceState state, + DataSourceStatus::ErrorInfo::ErrorKind kind, + std::string message); + + // Applies the leading cache initializers on the calling thread, so that + // cached flags are available as soon as Start() returns. A cache + // initializer that does not complete synchronously is left to the chain. + void RunCacheInitializers(); + + void RunNextInitializer(); + void OnInitializerResult(FDv2SourceResult result); + void StartSynchronizers(); + void RunSynchronizerNext(); + void OnSynchronizerResult(FDv2SourceResult result); + void OnConditionFired(IFDv2Condition::Type type); + + // Builds the conditions to apply to the currently active synchronizer. + // Must be called with mutex_ held. Reads source_manager_. + std::unique_ptr BuildActiveConditions() const; + + // Applies a changeset to the store and records what the result reported + // about the environment. + void ApplyResult(FDv2SourceResult::ChangeSet change_set, + std::optional environment_id, + bool from_cache); + + // Reports that no source can supply data, choosing the status that + // reflects why. + void ReportExhausted(bool any_synchronizers_configured); + + // Logger is itself thread-safe and cheap to copy. + Logger logger_; + + // Immutable after construction. + boost::asio::any_io_executor const executor_; + std::vector> const + initializer_factories_; + std::unique_ptr const fallback_condition_factory_; + std::unique_ptr const recovery_condition_factory_; + Context const context_; + // True when the cache is the only source that could ever supply data, in + // which case a cache miss still completes initialization successfully. + bool const cache_only_; + + // Non-owning. Lifetimes guaranteed by the caller (see constructor doc). + IDataSourceUpdateSink* const sink_; + flag_manager::FlagStore const* const store_; + DataSourceStatusManager* const status_manager_; + + // Set by Start() to detect repeat or concurrent calls. + std::atomic_bool start_called_; + + // Suppresses consecutive "interrupted" logs from the active synchronizer. + std::atomic_bool last_logged_synchronizer_interrupted_; + + // Orchestration state, protected by mutex_. + mutable std::mutex mutex_; + bool closed_; + bool received_data_; + std::optional environment_id_; + std::size_t initializer_index_; + // Whether active_initializer_ reads from the local cache. + bool active_initializer_from_cache_; + SourceManager source_manager_; + std::unique_ptr active_initializer_; + std::unique_ptr active_synchronizer_; + std::unique_ptr active_conditions_; +}; + +} // namespace launchdarkly::client_side::data_sources diff --git a/libs/client-sdk/tests/fdv2_data_source_test.cpp b/libs/client-sdk/tests/fdv2_data_source_test.cpp new file mode 100644 index 000000000..9309200b5 --- /dev/null +++ b/libs/client-sdk/tests/fdv2_data_source_test.cpp @@ -0,0 +1,768 @@ +#include + +#include +#include + +#include +#include +#include +#include +#include + +#include + +#include +#include +#include +#include +#include + +using namespace launchdarkly; +using namespace launchdarkly::client_side; +using namespace launchdarkly::client_side::data_sources; +using namespace std::chrono_literals; + +namespace { + +Logger MakeNullLogger() { + struct NullBackend : ILogBackend { + bool Enabled(LogLevel) noexcept override { return false; } + void Write(LogLevel, std::string) noexcept override {} + }; + return Logger{std::make_shared()}; +} + +// Initializer that resolves Run() with a single pre-set result. +class MockInitializer : public IFDv2Initializer { + public: + explicit MockInitializer(FDv2SourceResult result, + bool* closed_flag = nullptr) + : result_(std::move(result)), closed_flag_(closed_flag) {} + + async::Future Run() override { + return async::MakeFuture(std::move(result_)); + } + + void Close() override { + if (closed_flag_) { + *closed_flag_ = true; + } + } + + std::string const& Identity() const override { + static std::string const id = "mock initializer"; + return id; + } + + private: + FDv2SourceResult result_; + bool* closed_flag_; +}; + +// Synchronizer that resolves successive Next() calls from a queue of results. +// Once the queue is exhausted it returns Shutdown to end orchestration, unless +// stall_after_results is set, in which case the next Future never resolves. +class MockSynchronizer : public IFDv2Synchronizer { + public: + MockSynchronizer(std::vector results, + bool* closed_flag = nullptr, + std::vector* next_calls = nullptr, + bool stall_after_results = false) + : results_(std::move(results)), + closed_flag_(closed_flag), + next_calls_(next_calls), + stall_after_results_(stall_after_results) {} + + async::Future Next( + data_model::Selector selector) override { + if (next_calls_) { + next_calls_->push_back(selector); + } + if (call_index_ < results_.size()) { + return async::MakeFuture(std::move(results_[call_index_++])); + } + if (stall_after_results_) { + return stall_promise_.GetFuture(); + } + return async::MakeFuture( + FDv2SourceResult{FDv2SourceResult::Shutdown{}}); + } + + void Close() override { + stall_promise_.Resolve(FDv2SourceResult{FDv2SourceResult::Shutdown{}}); + if (closed_flag_) { + *closed_flag_ = true; + } + } + + std::string const& Identity() const override { + static std::string const id = "mock synchronizer"; + return id; + } + + private: + std::vector results_; + std::size_t call_index_ = 0; + bool* closed_flag_; + std::vector* next_calls_; + bool stall_after_results_; + async::Promise stall_promise_; +}; + +// Returns a pre-supplied source on its first Build() call. +class OneShotInitializerFactory : public IFDv2InitializerFactory { + public: + explicit OneShotInitializerFactory(std::unique_ptr source, + bool from_cache = false) + : source_(std::move(source)), from_cache_(from_cache) {} + + std::unique_ptr Build() override { + ++build_count_; + return std::move(source_); + } + + [[nodiscard]] bool IsFromCache() const override { return from_cache_; } + + int build_count_ = 0; + + private: + std::unique_ptr source_; + bool from_cache_; +}; + +class OneShotSynchronizerFactory : public IFDv2SynchronizerFactory { + public: + explicit OneShotSynchronizerFactory( + std::unique_ptr source) + : source_(std::move(source)) {} + + std::unique_ptr Build() override { + ++build_count_; + return std::move(source_); + } + + int build_count_ = 0; + + private: + std::unique_ptr source_; +}; + +// Returns each pre-supplied source in order on successive Build() calls, so +// that a factory reused by recovery can hand out a fresh source. +class MultiShotSynchronizerFactory : public IFDv2SynchronizerFactory { + public: + explicit MultiShotSynchronizerFactory( + std::vector> sources) + : sources_(std::move(sources)) {} + + std::unique_ptr Build() override { + ++build_count_; + if (build_count_ <= static_cast(sources_.size())) { + return std::move(sources_[build_count_ - 1]); + } + return nullptr; + } + + int build_count_ = 0; + + private: + std::vector> sources_; +}; + +// Initializer whose Run() never resolves, so that orchestration can be +// examined while it is in flight. +class StalledInitializer : public IFDv2Initializer { + public: + explicit StalledInitializer(bool* closed_flag) + : closed_flag_(closed_flag) {} + + async::Future Run() override { + return promise_.GetFuture(); + } + + void Close() override { + if (closed_flag_) { + *closed_flag_ = true; + } + } + + std::string const& Identity() const override { + static std::string const id = "stalled initializer"; + return id; + } + + private: + async::Promise promise_; + bool* closed_flag_; +}; + +data_model::Selector MakeSelector(std::int64_t version, std::string state) { + return data_model::Selector{ + data_model::Selector::State{version, std::move(state)}}; +} + +ItemDescriptor MakeFlag(std::uint64_t version, Value value) { + return ItemDescriptor{ + EvaluationResult{version, std::nullopt, false, false, std::nullopt, + EvaluationDetailInternal{std::move(value), + std::nullopt, std::nullopt}}}; +} + +FDv2SourceResult MakeChangeSetResult(data_model::ChangeSetType type, + FlagChangeSetData data, + data_model::Selector selector) { + return FDv2SourceResult{FDv2SourceResult::ChangeSet{ + FlagChangeSet{type, std::move(data), std::move(selector)}}}; +} + +FDv2SourceResult MakeErrorResult(FDv2SourceResult::Value value) { + return FDv2SourceResult{std::move(value)}; +} + +FDv2SourceResult::ErrorInfo MakeError(std::string message) { + return FDv2SourceResult::ErrorInfo{ + FDv2SourceResult::ErrorInfo::ErrorKind::kNetworkError, 0, + std::move(message), std::chrono::system_clock::now()}; +} + +// Records the from_cache flag of each apply before passing it along, so tests +// can assert how the data source classified its sources. +class RecordingSink : public IDataSourceUpdateSink { + public: + RecordingSink(IDataSourceUpdateSink* inner, std::vector* applies) + : inner_(inner), applies_(applies) {} + + void Init(Context const& context, + std::unordered_map data) override { + inner_->Init(context, std::move(data)); + } + + void Upsert(Context const& context, + std::string key, + ItemDescriptor item) override { + inner_->Upsert(context, std::move(key), std::move(item)); + } + + void Apply(Context const& context, + FlagChangeSet change_set, + bool from_cache) override { + applies_->push_back(from_cache); + inner_->Apply(context, std::move(change_set), from_cache); + } + + private: + IDataSourceUpdateSink* const inner_; + std::vector* const applies_; +}; + +// Owns everything a data source needs to run against a real flag store. +class Harness { + public: + Harness() + : flag_manager_("sdk-key", logger_, 5, nullptr), + sink_(&flag_manager_.Updater(), &applies_) {} + + std::shared_ptr MakeDataSource( + std::vector> initializers, + std::vector> synchronizers, + std::unique_ptr fallback = nullptr, + std::unique_ptr recovery = nullptr) { + return std::make_shared( + std::move(initializers), std::move(synchronizers), + std::move(fallback), std::move(recovery), ioc_.get_executor(), + ContextBuilder().Kind("user", "user-key").Build(), &sink_, + &flag_manager_.Store(), &status_manager_, logger_); + } + + boost::asio::io_context& Context() { return ioc_; } + DataSourceStatusManager& StatusManager() { return status_manager_; } + flag_manager::FlagStore const& Store() { return flag_manager_.Store(); } + + DataSourceStatus::DataSourceState State() { + return status_manager_.Status().State(); + } + + std::vector const& Applies() const { return applies_; } + + private: + Logger logger_ = MakeNullLogger(); + boost::asio::io_context ioc_; + DataSourceStatusManager status_manager_; + flag_manager::FlagManager flag_manager_; + std::vector applies_; + RecordingSink sink_; +}; + +} // namespace + +// ============================================================================ +// Lifecycle +// ============================================================================ + +TEST(ClientFDv2DataSourceTest, NoSourcesConfiguredIsImmediatelyValid) { + Harness h; + auto source = h.MakeDataSource({}, {}); + + source->Start(); + + EXPECT_EQ(DataSourceStatus::DataSourceState::kValid, h.State()); +} + +// The client drops a data source as soon as its replacement starts, so a +// status transition from the old one would land on top of the new one's. +TEST(ClientFDv2DataSourceTest, NoStatusIsReportedAfterShutdown) { + Harness h; + + std::vector> initializers; + initializers.push_back(std::make_unique( + std::make_unique(MakeChangeSetResult( + data_model::ChangeSetType::kNone, {}, data_model::Selector{})))); + + auto source = h.MakeDataSource(std::move(initializers), {}); + source->Start(); + + bool completed = false; + source->ShutdownAsync([&completed]() { completed = true; }); + h.StatusManager().SetState(DataSourceStatus::DataSourceState::kValid); + + // Whatever the abandoned orchestration had queued must not overwrite the + // state the caller sees after the shutdown. + h.Context().poll(); + + EXPECT_TRUE(completed); + EXPECT_EQ(DataSourceStatus::DataSourceState::kValid, h.State()); +} + +TEST(ClientFDv2DataSourceTest, ShutdownClosesTheActiveInitializer) { + Harness h; + bool closed = false; + + std::vector> initializers; + initializers.push_back(std::make_unique( + std::make_unique(&closed))); + + auto source = h.MakeDataSource(std::move(initializers), {}); + source->Start(); + h.Context().run(); + + bool completed = false; + source->ShutdownAsync([&completed] { completed = true; }); + h.Context().restart(); + h.Context().run(); + + EXPECT_TRUE(closed); + EXPECT_TRUE(completed); +} + +// ============================================================================ +// Initializer phase +// ============================================================================ + +TEST(ClientFDv2DataSourceTest, InitializerWithABasisAppliesAndBecomesValid) { + Harness h; + + std::vector> initializers; + initializers.push_back(std::make_unique( + std::make_unique( + MakeChangeSetResult(data_model::ChangeSetType::kFull, + {FlagChange{"flagA", MakeFlag(1, Value("a"))}}, + MakeSelector(1, "state-1"))))); + + auto source = h.MakeDataSource(std::move(initializers), {}); + source->Start(); + h.Context().run(); + + EXPECT_EQ(DataSourceStatus::DataSourceState::kValid, h.State()); + ASSERT_TRUE(h.Store().Get("flagA")); + EXPECT_EQ(Value("a"), h.Store().Get("flagA")->item->Detail().Value()); + ASSERT_TRUE(h.Store().CurrentSelector().value.has_value()); + EXPECT_EQ("state-1", h.Store().CurrentSelector().value->state); +} + +// An initializer that supplies data without a selector has not established a +// basis, so the chain keeps going to find one. +TEST(ClientFDv2DataSourceTest, DataWithoutASelectorContinuesTheChain) { + Harness h; + + std::vector> initializers; + initializers.push_back(std::make_unique( + std::make_unique(MakeChangeSetResult( + data_model::ChangeSetType::kFull, + {FlagChange{"cached", MakeFlag(1, Value("from-cache"))}}, + data_model::Selector{})))); + auto second = std::make_unique( + std::make_unique(MakeChangeSetResult( + data_model::ChangeSetType::kFull, + {FlagChange{"live", MakeFlag(1, Value("from-network"))}}, + MakeSelector(1, "state-1")))); + auto* second_ptr = second.get(); + initializers.push_back(std::move(second)); + + auto source = h.MakeDataSource(std::move(initializers), {}); + source->Start(); + h.Context().run(); + + EXPECT_EQ(1, second_ptr->build_count_); + EXPECT_FALSE(h.Store().Get("cached")); + ASSERT_TRUE(h.Store().Get("live")); +} + +TEST(ClientFDv2DataSourceTest, FailedInitializerAdvancesToTheNext) { + Harness h; + + std::vector> initializers; + initializers.push_back(std::make_unique( + std::make_unique(MakeErrorResult( + FDv2SourceResult::Interrupted{MakeError("boom")})))); + initializers.push_back(std::make_unique( + std::make_unique( + MakeChangeSetResult(data_model::ChangeSetType::kFull, + {FlagChange{"flagA", MakeFlag(1, Value("a"))}}, + MakeSelector(1, "state-1"))))); + + auto source = h.MakeDataSource(std::move(initializers), {}); + source->Start(); + h.Context().run(); + + EXPECT_EQ(DataSourceStatus::DataSourceState::kValid, h.State()); + EXPECT_TRUE(h.Store().Get("flagA")); +} + +TEST(ClientFDv2DataSourceTest, ExhaustedInitializersWithNoDataShutDown) { + Harness h; + + std::vector> initializers; + initializers.push_back(std::make_unique( + std::make_unique(MakeErrorResult( + FDv2SourceResult::TerminalError{MakeError("boom")})))); + + auto source = h.MakeDataSource(std::move(initializers), {}); + source->Start(); + h.Context().run(); + + EXPECT_EQ(DataSourceStatus::DataSourceState::kShutdown, h.State()); +} + +// Under FDv1 the client loaded the cache in its constructor, so cached flags +// were evaluable immediately. They still are. +TEST(ClientFDv2DataSourceTest, CachedDataIsAppliedBeforeStartReturns) { + Harness h; + + std::vector> initializers; + initializers.push_back(std::make_unique( + std::make_unique(MakeChangeSetResult( + data_model::ChangeSetType::kFull, + {FlagChange{"flagA", MakeFlag(1, Value("cached"))}}, + data_model::Selector{})), + /* from_cache= */ true)); + + auto source = h.MakeDataSource(std::move(initializers), {}); + source->Start(); + + // The executor has not run, so only the inline cache pass can have + // applied anything. + auto const flag = h.Store().Get("flagA"); + ASSERT_TRUE(flag); + EXPECT_EQ(Value("cached"), flag->item->Detail().Value()); + EXPECT_EQ(DataSourceStatus::DataSourceState::kValid, h.State()); +} + +// In offline mode the cache is the only thing that could ever supply data, +// so a miss means zero flags rather than a failure to start. +TEST(ClientFDv2DataSourceTest, CacheOnlyModeIsValidEvenOnAMiss) { + Harness h; + + std::vector> initializers; + initializers.push_back(std::make_unique( + std::make_unique(MakeChangeSetResult( + data_model::ChangeSetType::kNone, {}, data_model::Selector{})), + /* from_cache= */ true)); + + auto source = h.MakeDataSource(std::move(initializers), {}); + source->Start(); + h.Context().run(); + + EXPECT_EQ(DataSourceStatus::DataSourceState::kValid, h.State()); + EXPECT_TRUE(h.Store().GetAll().empty()); +} + +// A non-cache initializer returning "none" must not make initialization +// succeed when nothing else can supply data. +TEST(ClientFDv2DataSourceTest, NetworkOnlyNoneResultDoesNotCountAsSuccess) { + Harness h; + + std::vector> initializers; + initializers.push_back(std::make_unique( + std::make_unique(MakeChangeSetResult( + data_model::ChangeSetType::kNone, {}, data_model::Selector{})))); + + auto source = h.MakeDataSource(std::move(initializers), {}); + source->Start(); + h.Context().run(); + + EXPECT_EQ(DataSourceStatus::DataSourceState::kShutdown, h.State()); +} + +// ============================================================================ +// Synchronizer phase +// ============================================================================ + +TEST(ClientFDv2DataSourceTest, SynchronizerChangeSetsAreApplied) { + Harness h; + + std::vector results; + results.push_back( + MakeChangeSetResult(data_model::ChangeSetType::kFull, + {FlagChange{"flagA", MakeFlag(1, Value("a"))}}, + MakeSelector(1, "state-1"))); + results.push_back( + MakeChangeSetResult(data_model::ChangeSetType::kPartial, + {FlagChange{"flagA", MakeFlag(2, Value("a2"))}}, + MakeSelector(2, "state-2"))); + + std::vector> synchronizers; + synchronizers.push_back(std::make_unique( + std::make_unique(std::move(results)))); + + auto source = h.MakeDataSource({}, std::move(synchronizers)); + source->Start(); + h.Context().run(); + + ASSERT_TRUE(h.Store().Get("flagA")); + EXPECT_EQ(Value("a2"), h.Store().Get("flagA")->item->Detail().Value()); +} + +// The synchronizer asks the service for changes since the data the store +// already holds. +TEST(ClientFDv2DataSourceTest, SynchronizerReceivesTheStoresSelector) { + Harness h; + std::vector next_calls; + + std::vector results; + results.push_back( + MakeChangeSetResult(data_model::ChangeSetType::kFull, + {FlagChange{"flagA", MakeFlag(1, Value("a"))}}, + MakeSelector(1, "state-1"))); + + std::vector> synchronizers; + synchronizers.push_back(std::make_unique( + std::make_unique(std::move(results), nullptr, + &next_calls))); + + auto source = h.MakeDataSource({}, std::move(synchronizers)); + source->Start(); + h.Context().run(); + + ASSERT_GE(next_calls.size(), 2u); + EXPECT_FALSE(next_calls[0].value.has_value()); + ASSERT_TRUE(next_calls[1].value.has_value()); + EXPECT_EQ("state-1", next_calls[1].value->state); +} + +TEST(ClientFDv2DataSourceTest, InterruptedSynchronizerKeepsRunning) { + Harness h; + + std::vector results; + results.push_back( + MakeErrorResult(FDv2SourceResult::Interrupted{MakeError("boom")})); + results.push_back( + MakeChangeSetResult(data_model::ChangeSetType::kFull, + {FlagChange{"flagA", MakeFlag(1, Value("a"))}}, + MakeSelector(1, "state-1"))); + + std::vector> synchronizers; + synchronizers.push_back(std::make_unique( + std::make_unique(std::move(results)))); + + auto source = h.MakeDataSource({}, std::move(synchronizers)); + source->Start(); + h.Context().run(); + + EXPECT_EQ(DataSourceStatus::DataSourceState::kValid, h.State()); + EXPECT_TRUE(h.Store().Get("flagA")); +} + +TEST(ClientFDv2DataSourceTest, TerminalErrorAdvancesToTheNextSynchronizer) { + Harness h; + + std::vector first_results; + first_results.push_back( + MakeErrorResult(FDv2SourceResult::TerminalError{MakeError("gone")})); + + std::vector second_results; + second_results.push_back( + MakeChangeSetResult(data_model::ChangeSetType::kFull, + {FlagChange{"flagA", MakeFlag(1, Value("a"))}}, + MakeSelector(1, "state-1"))); + + std::vector> synchronizers; + synchronizers.push_back(std::make_unique( + std::make_unique(std::move(first_results)))); + auto second = std::make_unique( + std::make_unique(std::move(second_results))); + auto* second_ptr = second.get(); + synchronizers.push_back(std::move(second)); + + auto source = h.MakeDataSource({}, std::move(synchronizers)); + source->Start(); + h.Context().run(); + + EXPECT_EQ(1, second_ptr->build_count_); + EXPECT_TRUE(h.Store().Get("flagA")); +} + +TEST(ClientFDv2DataSourceTest, ExhaustedSynchronizersShutDown) { + Harness h; + + std::vector results; + results.push_back( + MakeErrorResult(FDv2SourceResult::TerminalError{MakeError("gone")})); + + std::vector> synchronizers; + synchronizers.push_back(std::make_unique( + std::make_unique(std::move(results)))); + + auto source = h.MakeDataSource({}, std::move(synchronizers)); + source->Start(); + h.Context().run(); + + EXPECT_EQ(DataSourceStatus::DataSourceState::kShutdown, h.State()); +} + +// ============================================================================ +// Environment ID +// ============================================================================ + +TEST(ClientFDv2DataSourceTest, RecordsTheEnvironmentIdFromAResult) { + Harness h; + + auto result = + MakeChangeSetResult(data_model::ChangeSetType::kFull, + {FlagChange{"flagA", MakeFlag(1, Value("a"))}}, + MakeSelector(1, "state-1")); + result.environment_id = "env-1234"; + + std::vector> initializers; + initializers.push_back(std::make_unique( + std::make_unique(std::move(result)))); + + auto source = h.MakeDataSource(std::move(initializers), {}); + source->Start(); + h.Context().run(); + + ASSERT_TRUE(source->EnvironmentId().has_value()); + EXPECT_EQ("env-1234", *source->EnvironmentId()); +} + +TEST(ClientFDv2DataSourceTest, ReportsNoEnvironmentIdUntilOneArrives) { + Harness h; + auto source = h.MakeDataSource({}, {}); + + source->Start(); + + EXPECT_FALSE(source->EnvironmentId().has_value()); +} + +// ============================================================================ +// Fallback and recovery +// ============================================================================ + +TEST(ClientFDv2DataSourceTest, SustainedInterruptionFallsBackToTheNextTier) { + Harness h; + + std::vector first_results; + first_results.push_back( + MakeErrorResult(FDv2SourceResult::Interrupted{MakeError("flaky")})); + + std::vector second_results; + second_results.push_back( + MakeChangeSetResult(data_model::ChangeSetType::kFull, + {FlagChange{"flagA", MakeFlag(1, Value("a"))}}, + MakeSelector(1, "state-1"))); + + std::vector> synchronizers; + synchronizers.push_back(std::make_unique( + std::make_unique(std::move(first_results), nullptr, + nullptr, + /* stall_after_results= */ true))); + auto second = std::make_unique( + std::make_unique(std::move(second_results))); + auto* second_ptr = second.get(); + synchronizers.push_back(std::move(second)); + + auto source = h.MakeDataSource({}, std::move(synchronizers), + std::make_unique( + h.Context().get_executor(), 50ms)); + source->Start(); + h.Context().run(); + + EXPECT_EQ(1, second_ptr->build_count_); + EXPECT_TRUE(h.Store().Get("flagA")); +} + +TEST(ClientFDv2DataSourceTest, RecoveryReturnsToThePreferredTier) { + Harness h; + + // Interrupt, then stall so that the fallback condition wins the race. + std::vector first_attempt; + first_attempt.push_back( + MakeErrorResult(FDv2SourceResult::Interrupted{MakeError("flaky")})); + + std::vector> preferred_sources; + preferred_sources.push_back(std::make_unique( + std::move(first_attempt), nullptr, nullptr, + /* stall_after_results= */ true)); + preferred_sources.push_back( + std::make_unique(std::vector{})); + + std::vector> synchronizers; + auto preferred = std::make_unique( + std::move(preferred_sources)); + auto* preferred_ptr = preferred.get(); + synchronizers.push_back(std::move(preferred)); + synchronizers.push_back(std::make_unique( + std::make_unique(std::vector{}, + nullptr, nullptr, + /* stall_after_results= */ true))); + + auto source = h.MakeDataSource({}, std::move(synchronizers), + std::make_unique( + h.Context().get_executor(), 50ms), + std::make_unique( + h.Context().get_executor(), 50ms)); + source->Start(); + h.Context().run(); + + EXPECT_EQ(2, preferred_ptr->build_count_); +} + +// ============================================================================ +// Cache-sourced data +// ============================================================================ + +// The store needs to know which data came from the cache, so that it is not +// written straight back to the cache it was read from. +TEST(ClientFDv2DataSourceTest, CacheSourcedDataIsMarkedAsSuch) { + Harness h; + + std::vector> initializers; + initializers.push_back(std::make_unique( + std::make_unique( + MakeChangeSetResult(data_model::ChangeSetType::kFull, + {FlagChange{"cached", MakeFlag(1, Value("a"))}}, + data_model::Selector{})), + /* from_cache= */ true)); + initializers.push_back(std::make_unique( + std::make_unique( + MakeChangeSetResult(data_model::ChangeSetType::kFull, + {FlagChange{"live", MakeFlag(1, Value("b"))}}, + MakeSelector(1, "state-1"))))); + + auto source = h.MakeDataSource(std::move(initializers), {}); + source->Start(); + h.Context().run(); + + ASSERT_EQ(2u, h.Applies().size()); + EXPECT_TRUE(h.Applies()[0]); + EXPECT_FALSE(h.Applies()[1]); +} diff --git a/libs/server-sdk/src/data_systems/fdv2/conditions.hpp b/libs/internal/include/launchdarkly/data_sources/fdv2/conditions.hpp similarity index 70% rename from libs/server-sdk/src/data_systems/fdv2/conditions.hpp rename to libs/internal/include/launchdarkly/data_sources/fdv2/conditions.hpp index 70a3518f7..56a3eeea1 100644 --- a/libs/server-sdk/src/data_systems/fdv2/conditions.hpp +++ b/libs/internal/include/launchdarkly/data_sources/fdv2/conditions.hpp @@ -1,6 +1,6 @@ #pragma once -#include "../../data_interfaces/source/ifdv2_condition.hpp" +#include #include #include @@ -14,7 +14,7 @@ #include #include -namespace launchdarkly::server_side::data_systems { +namespace launchdarkly::internal::data_sources { /** * Base class for conditions that fire after a duration elapses on the @@ -25,8 +25,12 @@ namespace launchdarkly::server_side::data_systems { * Derived classes implement Inform() to translate orchestrator events into * arm/cancel actions on the timer. Subclasses also implement GetType() to * report whether they are a fallback or recovery condition. + * + * Thread-safe: every method may be called from any thread. The timer state is + * held behind a mutex in a shared State, so a timer callback firing on the + * executor is safe against a concurrent Close() from a caller's thread. */ -class TimedCondition : public data_interfaces::IFDv2Condition { +class TimedCondition : public IFDv2Condition { public: TimedCondition(boost::asio::any_io_executor executor, std::chrono::milliseconds timeout); @@ -54,6 +58,9 @@ class TimedCondition : public data_interfaces::IFDv2Condition { private: struct State { std::mutex mutex; + // All protected by mutex. timer_cancel is replaced when the timer is + // re-armed, so the lock covers the replacement and not just the + // source's own operations. bool closed = false; async::Promise promise; std::optional timer_cancel; @@ -68,13 +75,15 @@ class TimedCondition : public data_interfaces::IFDv2Condition { * Fires after the active synchronizer has been continuously interrupted for * the configured timeout. Each CHANGE_SET result cancels any pending timer; * the next Interrupted status re-arms it. + * + * Thread-safe, as TimedCondition is. */ class FallbackCondition final : public TimedCondition { public: FallbackCondition(boost::asio::any_io_executor executor, std::chrono::milliseconds timeout); - void Inform(data_interfaces::FDv2SourceResult const& result) override; + void Inform(SourceSignal signal) override; [[nodiscard]] Type GetType() const override { return Type::kFallback; } }; @@ -83,31 +92,33 @@ class FallbackCondition final : public TimedCondition { * Fires after the active synchronizer has been running for the configured * timeout, regardless of result content. The timer is started at * construction; Inform() is a no-op. + * + * Thread-safe, as TimedCondition is. */ class RecoveryCondition final : public TimedCondition { public: RecoveryCondition(boost::asio::any_io_executor executor, std::chrono::milliseconds timeout); - void Inform(data_interfaces::FDv2SourceResult const& result) override; + void Inform(SourceSignal signal) override; [[nodiscard]] Type GetType() const override { return Type::kRecovery; } }; /** * Builds fresh FallbackCondition instances on demand. + * + * Thread-safe: Build() and GetType() may be called from any thread, and + * hold no state beyond the executor and timeout given at construction. */ -class FallbackConditionFactory final - : public data_interfaces::IFDv2ConditionFactory { +class FallbackConditionFactory final : public IFDv2ConditionFactory { public: FallbackConditionFactory(boost::asio::any_io_executor executor, std::chrono::milliseconds timeout); - [[nodiscard]] std::unique_ptr Build() - override; + [[nodiscard]] std::unique_ptr Build() override; - [[nodiscard]] data_interfaces::IFDv2Condition::Type GetType() - const override; + [[nodiscard]] IFDv2Condition::Type GetType() const override; private: boost::asio::any_io_executor const executor_; @@ -116,18 +127,18 @@ class FallbackConditionFactory final /** * Builds fresh RecoveryCondition instances on demand. + * + * Thread-safe: Build() and GetType() may be called from any thread, and + * hold no state beyond the executor and timeout given at construction. */ -class RecoveryConditionFactory final - : public data_interfaces::IFDv2ConditionFactory { +class RecoveryConditionFactory final : public IFDv2ConditionFactory { public: RecoveryConditionFactory(boost::asio::any_io_executor executor, std::chrono::milliseconds timeout); - [[nodiscard]] std::unique_ptr Build() - override; + [[nodiscard]] std::unique_ptr Build() override; - [[nodiscard]] data_interfaces::IFDv2Condition::Type GetType() - const override; + [[nodiscard]] IFDv2Condition::Type GetType() const override; private: boost::asio::any_io_executor const executor_; @@ -145,8 +156,7 @@ class RecoveryConditionFactory final class Conditions final { public: explicit Conditions( - std::vector> - conditions); + std::vector> conditions); ~Conditions(); @@ -161,29 +171,30 @@ class Conditions final { * `token` once the result is no longer needed, so that the per-call * Promise (and its registered continuations) can be released. */ - [[nodiscard]] async::Future - GetFuture(async::CancellationToken token); + [[nodiscard]] async::Future GetFuture( + async::CancellationToken token); - void Inform(data_interfaces::FDv2SourceResult const& result); + void Inform(SourceSignal signal); void Close(); private: struct PendingEntry { std::int64_t id; - async::Promise promise; + async::Promise promise; std::unique_ptr cancel_cb; }; struct State { std::mutex mutex; + // All protected by mutex. std::int64_t next_id = 0; std::vector pending; - std::optional aggregate_result; + std::optional aggregate_result; }; - std::vector> conditions_; + std::vector> conditions_; std::shared_ptr const state_; }; -} // namespace launchdarkly::server_side::data_systems +} // namespace launchdarkly::internal::data_sources diff --git a/libs/server-sdk/src/data_interfaces/source/ifdv2_condition.hpp b/libs/internal/include/launchdarkly/data_sources/fdv2/ifdv2_condition.hpp similarity index 82% rename from libs/server-sdk/src/data_interfaces/source/ifdv2_condition.hpp rename to libs/internal/include/launchdarkly/data_sources/fdv2/ifdv2_condition.hpp index ea19f0520..be19a7d01 100644 --- a/libs/server-sdk/src/data_interfaces/source/ifdv2_condition.hpp +++ b/libs/internal/include/launchdarkly/data_sources/fdv2/ifdv2_condition.hpp @@ -1,15 +1,27 @@ #pragma once -#include "fdv2_source_result.hpp" - #include #include -namespace launchdarkly::server_side::data_interfaces { +namespace launchdarkly::internal::data_sources { + +/** + * What the orchestrator observed from the active synchronizer, reduced to the + * distinctions a condition acts on. The orchestrator maps its own result type + * onto this before informing its conditions. + */ +enum class SourceSignal { + /** A changeset arrived. */ + kChangeSet, + /** The synchronizer reported a recoverable failure. */ + kInterrupted, + /** Anything else, which no condition acts on. */ + kOther, +}; /** - * A condition observes the orchestrator's stream of synchronizer results and + * A condition observes the orchestrator's stream of synchronizer signals and * fires when criteria for a synchronizer transition are met. * * Each condition plays one of two roles, identified by Type(): @@ -18,7 +30,7 @@ namespace launchdarkly::server_side::data_interfaces { * - kRecovery: when fired, the orchestrator stops the active fallback * synchronizer and returns to the most-preferred synchronizer. * - * Conditions are stateful: the orchestrator pushes results into a condition + * Conditions are stateful: the orchestrator pushes signals into a condition * via Inform() so the condition can update its internal state (typically a * timer). When the condition's criteria are satisfied, the future returned * by Execute() resolves with the condition's Type. @@ -52,10 +64,10 @@ class IFDv2Condition { [[nodiscard]] virtual async::Future Execute() = 0; /** - * Pushes a synchronizer result into the condition so it can update any + * Pushes a synchronizer signal into the condition so it can update any * internal state (e.g., arm or cancel a timer). */ - virtual void Inform(FDv2SourceResult const& result) = 0; + virtual void Inform(SourceSignal signal) = 0; /** * Cancels any pending internal work and resolves the future returned by @@ -104,4 +116,4 @@ class IFDv2ConditionFactory { IFDv2ConditionFactory() = default; }; -} // namespace launchdarkly::server_side::data_interfaces +} // namespace launchdarkly::internal::data_sources diff --git a/libs/server-sdk/src/data_systems/fdv2/source_manager.hpp b/libs/internal/include/launchdarkly/data_sources/fdv2/source_manager.hpp similarity index 52% rename from libs/server-sdk/src/data_systems/fdv2/source_manager.hpp rename to libs/internal/include/launchdarkly/data_sources/fdv2/source_manager.hpp index 264dfd494..5d6d1b42b 100644 --- a/libs/server-sdk/src/data_systems/fdv2/source_manager.hpp +++ b/libs/internal/include/launchdarkly/data_sources/fdv2/source_manager.hpp @@ -1,13 +1,11 @@ #pragma once -#include "../../data_interfaces/source/ifdv2_synchronizer.hpp" -#include "../../data_interfaces/source/ifdv2_synchronizer_factory.hpp" - #include #include +#include #include -namespace launchdarkly::server_side::data_systems { +namespace launchdarkly::internal::data_sources { /** * Manages a list of synchronizer factories together with per-factory state @@ -26,12 +24,25 @@ namespace launchdarkly::server_side::data_systems { * Factories whose IsFDv1Fallback() returns true start in the Blocked state. * * Not thread-safe. The caller is responsible for serializing all calls. + * + * @tparam Factory The SDK's synchronizer factory interface, which must supply + * IsFDv1Fallback() and a Build() returning a smart pointer to a synchronizer. */ +template class SourceManager { public: - explicit SourceManager( - std::vector> - factories); + using SynchronizerPtr = decltype(std::declval().Build()); + + explicit SourceManager(std::vector> factories) { + synchronizers_.reserve(factories.size()); + for (auto& factory : factories) { + bool const is_fdv1_fallback = factory->IsFDv1Fallback(); + synchronizers_.push_back(SynchronizerFactoryWithState{ + std::move(factory), + is_fdv1_fallback ? State::kBlocked : State::kAvailable, + is_fdv1_fallback}); + } + } /** * Advances to the next Available synchronizer factory (wrapping past the @@ -39,19 +50,40 @@ class SourceManager { * as the current one for subsequent queries. Returns nullptr if no * Available factory exists. */ - std::unique_ptr NextSynchronizer(); + SynchronizerPtr NextSynchronizer() { + if (synchronizers_.empty()) { + current_factory_index_ = -1; + return nullptr; + } + for (std::size_t visited = 0; visited < synchronizers_.size(); + ++visited) { + synchronizer_index_ = (synchronizer_index_ + 1) % + static_cast(synchronizers_.size()); + if (synchronizers_[synchronizer_index_].state == + State::kAvailable) { + current_factory_index_ = synchronizer_index_; + return synchronizers_[synchronizer_index_].factory->Build(); + } + } + current_factory_index_ = -1; + return nullptr; + } /** * Marks the currently tracked factory as Blocked. No-op if no factory is * currently tracked. */ - void BlockCurrentSynchronizer(); + void BlockCurrentSynchronizer() { + if (current_factory_index_ >= 0) { + synchronizers_[current_factory_index_].state = State::kBlocked; + } + } /** * Resets the iteration cursor so that the next call to NextSynchronizer * begins searching from index 0. */ - void ResetSourceIndex(); + void ResetSourceIndex() { synchronizer_index_ = -1; } /** * Blocks every non-FDv1 factory and unblocks the FDv1 fallback factory, @@ -59,37 +91,69 @@ class SourceManager { * NextSynchronizer returns the FDv1 fallback. If no FDv1 fallback factory * was configured, every factory is left blocked. */ - void SwitchToFDv1Fallback(); + void SwitchToFDv1Fallback() { + for (auto& entry : synchronizers_) { + entry.state = + entry.is_fdv1_fallback ? State::kAvailable : State::kBlocked; + } + synchronizer_index_ = -1; + } /** * Returns synchronizer state to the initial configuration, including * unblocking factories previously blocked by terminal errors. */ - void SwitchBackToFDv2(); + void SwitchBackToFDv2() { + for (auto& entry : synchronizers_) { + entry.state = + entry.is_fdv1_fallback ? State::kBlocked : State::kAvailable; + } + synchronizer_index_ = -1; + } /** * Returns true if the currently tracked factory is the first Available * factory in the list. Returns false if no factory is currently tracked. */ - [[nodiscard]] bool IsPrimeSynchronizer() const; + [[nodiscard]] bool IsPrimeSynchronizer() const { + for (std::size_t i = 0; i < synchronizers_.size(); ++i) { + if (synchronizers_[i].state == State::kAvailable) { + return synchronizer_index_ == static_cast(i); + } + } + return false; + } /** * Returns the count of factories not in the Blocked state. */ - [[nodiscard]] std::size_t AvailableSynchronizerCount() const; + [[nodiscard]] std::size_t AvailableSynchronizerCount() const { + std::size_t count = 0; + for (auto const& s : synchronizers_) { + if (s.state == State::kAvailable) { + ++count; + } + } + return count; + } /** * Returns the total number of factories configured at construction * (including any currently in the Blocked state). Constant for the * lifetime of the SourceManager. */ - [[nodiscard]] std::size_t SynchronizerCount() const; + [[nodiscard]] std::size_t SynchronizerCount() const { + return synchronizers_.size(); + } /** * Returns true if the currently tracked factory is the FDv1 fallback * synchronizer. */ - [[nodiscard]] bool IsCurrentSynchronizerFDv1Fallback() const; + [[nodiscard]] bool IsCurrentSynchronizerFDv1Fallback() const { + return current_factory_index_ >= 0 && + synchronizers_[current_factory_index_].is_fdv1_fallback; + } SourceManager(SourceManager const&) = delete; SourceManager(SourceManager&&) = delete; @@ -101,7 +165,7 @@ class SourceManager { enum class State { kAvailable, kBlocked }; struct SynchronizerFactoryWithState { - std::unique_ptr factory; + std::unique_ptr factory; State state = State::kAvailable; bool is_fdv1_fallback = false; }; @@ -113,4 +177,4 @@ class SourceManager { int current_factory_index_ = -1; }; -} // namespace launchdarkly::server_side::data_systems +} // namespace launchdarkly::internal::data_sources diff --git a/libs/internal/src/CMakeLists.txt b/libs/internal/src/CMakeLists.txt index 1b3d54e3c..40ccbded2 100644 --- a/libs/internal/src/CMakeLists.txt +++ b/libs/internal/src/CMakeLists.txt @@ -8,12 +8,14 @@ file(GLOB HEADER_LIST CONFIGURE_DEPENDS "${LaunchDarklyInternalSdk_SOURCE_DIR}/include/launchdarkly/serialization/events/*.hpp" "${LaunchDarklyInternalSdk_SOURCE_DIR}/include/launchdarkly/signals/*.hpp" "${LaunchDarklyInternalSdk_SOURCE_DIR}/include/launchdarkly/data_sources/*.hpp" + "${LaunchDarklyInternalSdk_SOURCE_DIR}/include/launchdarkly/data_sources/fdv2/*.hpp" ) # Automatic library: static or dynamic based on user config. set(INTERNAL_SOURCES ${HEADER_LIST} context_filter.cpp + data_sources/fdv2/conditions.cpp events/asio_event_processor.cpp events/null_event_processor.cpp events/common_events.cpp diff --git a/libs/server-sdk/src/data_systems/fdv2/conditions.cpp b/libs/internal/src/data_sources/fdv2/conditions.cpp similarity index 92% rename from libs/server-sdk/src/data_systems/fdv2/conditions.cpp rename to libs/internal/src/data_sources/fdv2/conditions.cpp index 01350eb93..3c9e9a09b 100644 --- a/libs/server-sdk/src/data_systems/fdv2/conditions.cpp +++ b/libs/internal/src/data_sources/fdv2/conditions.cpp @@ -1,15 +1,11 @@ -#include "conditions.hpp" +#include #include #include #include -#include -namespace launchdarkly::server_side::data_systems { - -using data_interfaces::FDv2SourceResult; -using data_interfaces::IFDv2Condition; +namespace launchdarkly::internal::data_sources { TimedCondition::TimedCondition(boost::asio::any_io_executor executor, std::chrono::milliseconds timeout) @@ -73,10 +69,10 @@ FallbackCondition::FallbackCondition(boost::asio::any_io_executor executor, std::chrono::milliseconds timeout) : TimedCondition(std::move(executor), timeout) {} -void FallbackCondition::Inform(FDv2SourceResult const& result) { - if (std::get_if(&result.value)) { +void FallbackCondition::Inform(SourceSignal signal) { + if (signal == SourceSignal::kChangeSet) { CancelTimer(); - } else if (std::get_if(&result.value)) { + } else if (signal == SourceSignal::kInterrupted) { ArmTimer(); } } @@ -87,7 +83,7 @@ RecoveryCondition::RecoveryCondition(boost::asio::any_io_executor executor, ArmTimer(); } -void RecoveryCondition::Inform(FDv2SourceResult const&) {} +void RecoveryCondition::Inform(SourceSignal) {} FallbackConditionFactory::FallbackConditionFactory( boost::asio::any_io_executor executor, @@ -211,9 +207,9 @@ async::Future Conditions::GetFuture( return future; } -void Conditions::Inform(FDv2SourceResult const& result) { +void Conditions::Inform(SourceSignal signal) { for (auto const& condition : conditions_) { - condition->Inform(result); + condition->Inform(signal); } } @@ -235,4 +231,4 @@ void Conditions::Close() { } } -} // namespace launchdarkly::server_side::data_systems +} // namespace launchdarkly::internal::data_sources diff --git a/libs/server-sdk/tests/conditions_test.cpp b/libs/internal/tests/conditions_test.cpp similarity index 76% rename from libs/server-sdk/tests/conditions_test.cpp rename to libs/internal/tests/conditions_test.cpp index 487833cb7..2466635d1 100644 --- a/libs/server-sdk/tests/conditions_test.cpp +++ b/libs/internal/tests/conditions_test.cpp @@ -1,6 +1,6 @@ #include -#include +#include #include #include @@ -8,8 +8,7 @@ #include #include -using namespace launchdarkly::server_side::data_interfaces; -using namespace launchdarkly::server_side::data_systems; +using namespace launchdarkly::internal::data_sources; using namespace std::chrono_literals; using launchdarkly::async::CancellationToken; @@ -52,11 +51,7 @@ TEST(FallbackConditionTest, InterruptedArmsTimerWhichFiresAfterTimeout) { FallbackCondition condition(ioc.GetExecutor(), /*timeout=*/100ms); auto future = condition.Execute(); - condition.Inform(FDv2SourceResult{FDv2SourceResult::Interrupted{ - FDv2SourceResult::ErrorInfo{ - FDv2SourceResult::ErrorInfo::ErrorKind::kNetworkError, - /*status_code=*/0, "boom", std::chrono::system_clock::now()}, - }}); + condition.Inform(SourceSignal::kInterrupted); auto result = future.WaitForResult(1s); @@ -71,18 +66,8 @@ TEST(FallbackConditionTest, ChangeSetCancelsActiveTimer) { // Arm the timer with Interrupted, then cancel via ChangeSet before it // fires. - condition.Inform(FDv2SourceResult{FDv2SourceResult::Interrupted{ - FDv2SourceResult::ErrorInfo{ - FDv2SourceResult::ErrorInfo::ErrorKind::kNetworkError, - /*status_code=*/0, "boom", std::chrono::system_clock::now()}, - }}); - condition.Inform(FDv2SourceResult{FDv2SourceResult::ChangeSet{ - launchdarkly::data_model::ChangeSet{ - launchdarkly::data_model::ChangeSetType::kFull, - {}, - launchdarkly::data_model::Selector{}, - }, - }}); + condition.Inform(SourceSignal::kInterrupted); + condition.Inform(SourceSignal::kChangeSet); // Wait well past the 100ms threshold; future should remain unresolved. std::this_thread::sleep_for(300ms); @@ -94,11 +79,7 @@ TEST(FallbackConditionTest, CloseCancelsActiveTimerAndResolvesWithCancelled) { FallbackCondition condition(ioc.GetExecutor(), /*timeout=*/100ms); auto future = condition.Execute(); - condition.Inform(FDv2SourceResult{FDv2SourceResult::Interrupted{ - FDv2SourceResult::ErrorInfo{ - FDv2SourceResult::ErrorInfo::ErrorKind::kNetworkError, - /*status_code=*/0, "boom", std::chrono::system_clock::now()}, - }}); + condition.Inform(SourceSignal::kInterrupted); condition.Close(); auto result = future.WaitForResult(200ms); @@ -127,18 +108,8 @@ TEST(RecoveryConditionTest, InformDoesNotAffectTimer) { // Recovery is purely time-based; results from the synchronizer should not // disturb the timer in either direction. - condition.Inform(FDv2SourceResult{FDv2SourceResult::Interrupted{ - FDv2SourceResult::ErrorInfo{ - FDv2SourceResult::ErrorInfo::ErrorKind::kNetworkError, - /*status_code=*/0, "boom", std::chrono::system_clock::now()}, - }}); - condition.Inform(FDv2SourceResult{FDv2SourceResult::ChangeSet{ - launchdarkly::data_model::ChangeSet{ - launchdarkly::data_model::ChangeSetType::kFull, - {}, - launchdarkly::data_model::Selector{}, - }, - }}); + condition.Inform(SourceSignal::kInterrupted); + condition.Inform(SourceSignal::kChangeSet); auto result = future.WaitForResult(1s); ASSERT_TRUE(result.has_value()); @@ -198,11 +169,7 @@ TEST(ConditionsTest, InformForwardsToAllUnderlyingConditions) { std::make_unique(ioc.GetExecutor(), /*timeout=*/1s)); Conditions conditions(std::move(conds)); - conditions.Inform(FDv2SourceResult{FDv2SourceResult::Interrupted{ - FDv2SourceResult::ErrorInfo{ - FDv2SourceResult::ErrorInfo::ErrorKind::kNetworkError, - /*status_code=*/0, "boom", std::chrono::system_clock::now()}, - }}); + conditions.Inform(SourceSignal::kInterrupted); auto result = conditions.GetFuture(CancellationToken{}).WaitForResult(1s); diff --git a/libs/server-sdk/tests/source_manager_test.cpp b/libs/internal/tests/source_manager_test.cpp similarity index 83% rename from libs/server-sdk/tests/source_manager_test.cpp rename to libs/internal/tests/source_manager_test.cpp index 68bfd5cc2..a6af67e20 100644 --- a/libs/server-sdk/tests/source_manager_test.cpp +++ b/libs/internal/tests/source_manager_test.cpp @@ -1,41 +1,31 @@ #include -#include -#include -#include +#include #include #include #include #include -using namespace launchdarkly::server_side::data_interfaces; -using namespace launchdarkly::server_side::data_systems; - namespace { // Stub synchronizer; SourceManager only cares that Build() returns one. -class StubSynchronizer : public IFDv2Synchronizer { - public: - launchdarkly::async::Future Next( - launchdarkly::data_model::Selector) override { - return launchdarkly::async::MakeFuture( - FDv2SourceResult{FDv2SourceResult::Shutdown{}}); - } - - void Close() override {} +class StubSynchronizer {}; - std::string const& Identity() const override { - static std::string const id = "stub"; - return id; - } +// Stands in for an SDK's synchronizer factory interface, which is all +// SourceManager requires of its type parameter. +class StubFactory { + public: + virtual std::unique_ptr Build() = 0; + [[nodiscard]] virtual bool IsFDv1Fallback() const { return false; } + virtual ~StubFactory() = default; }; // Counts Build() calls for assertion. Tests don't run the returned // synchronizer, so a fresh stub each time is fine. -class CountingFactory : public IFDv2SynchronizerFactory { +class CountingFactory : public StubFactory { public: - std::unique_ptr Build() override { + std::unique_ptr Build() override { ++build_count; return std::make_unique(); } @@ -50,6 +40,9 @@ class FDv1FallbackFactory : public CountingFactory { } // namespace +using SourceManager = + launchdarkly::internal::data_sources::SourceManager; + TEST(SourceManagerTest, EmptyManagerReportsZeroAvailable) { SourceManager mgr({}); @@ -64,7 +57,7 @@ TEST(SourceManagerTest, NextSynchronizerReturnsFirstThenWrapsAround) { auto f1 = std::make_unique(); auto* f0_ptr = f0.get(); auto* f1_ptr = f1.get(); - std::vector> factories; + std::vector> factories; factories.push_back(std::move(f0)); factories.push_back(std::move(f1)); SourceManager mgr(std::move(factories)); @@ -88,7 +81,7 @@ TEST(SourceManagerTest, BlockCurrentSynchronizerRemovesItFromRotation) { auto* f0_ptr = f0.get(); auto* f1_ptr = f1.get(); auto* f2_ptr = f2.get(); - std::vector> factories; + std::vector> factories; factories.push_back(std::move(f0)); factories.push_back(std::move(f1)); factories.push_back(std::move(f2)); @@ -117,7 +110,7 @@ TEST(SourceManagerTest, BlockCurrentSynchronizerRemovesItFromRotation) { TEST(SourceManagerTest, AllBlockedReturnsNullAndZeroCount) { auto f0 = std::make_unique(); auto f1 = std::make_unique(); - std::vector> factories; + std::vector> factories; factories.push_back(std::move(f0)); factories.push_back(std::move(f1)); SourceManager mgr(std::move(factories)); @@ -138,7 +131,7 @@ TEST(SourceManagerTest, ResetSourceIndexSendsNextCallToTheFirstAvailable) { auto f2 = std::make_unique(); auto* f0_ptr = f0.get(); auto* f2_ptr = f2.get(); - std::vector> factories; + std::vector> factories; factories.push_back(std::move(f0)); factories.push_back(std::move(f1)); factories.push_back(std::move(f2)); @@ -162,7 +155,7 @@ TEST(SourceManagerTest, ResetSourceIndexSkipsBlockedFirstFactory) { auto f1 = std::make_unique(); auto* f0_ptr = f0.get(); auto* f1_ptr = f1.get(); - std::vector> factories; + std::vector> factories; factories.push_back(std::move(f0)); factories.push_back(std::move(f1)); SourceManager mgr(std::move(factories)); @@ -183,7 +176,7 @@ TEST(SourceManagerTest, ResetSourceIndexSkipsBlockedFirstFactory) { TEST(SourceManagerTest, IsCurrentSynchronizerFDv1FallbackFalseForFDv2Factory) { auto f0 = std::make_unique(); - std::vector> factories; + std::vector> factories; factories.push_back(std::move(f0)); SourceManager mgr(std::move(factories)); @@ -195,7 +188,7 @@ TEST(SourceManagerTest, FDv1FallbackFactoryStartsBlockedAndIsSkipped) { auto fdv2 = std::make_unique(); auto fdv1 = std::make_unique(); auto* fdv1_ptr = fdv1.get(); - std::vector> factories; + std::vector> factories; factories.push_back(std::move(fdv2)); factories.push_back(std::move(fdv1)); SourceManager mgr(std::move(factories)); @@ -210,7 +203,7 @@ TEST(SourceManagerTest, SwitchToFDv1FallbackBlocksFDv2AndUnblocksFDv1) { auto fdv2 = std::make_unique(); auto fdv1 = std::make_unique(); auto* fdv1_ptr = fdv1.get(); - std::vector> factories; + std::vector> factories; factories.push_back(std::move(fdv2)); factories.push_back(std::move(fdv1)); SourceManager mgr(std::move(factories)); @@ -226,7 +219,7 @@ TEST(SourceManagerTest, SwitchToFDv1FallbackBlocksFDv2AndUnblocksFDv1) { TEST(SourceManagerTest, SwitchToFDv1FallbackWithoutAdapterBlocksEverything) { auto fdv2 = std::make_unique(); - std::vector> factories; + std::vector> factories; factories.push_back(std::move(fdv2)); SourceManager mgr(std::move(factories)); @@ -240,7 +233,7 @@ TEST(SourceManagerTest, SwitchToFDv1FallbackUnblocksPreviouslyBlockedFDv2) { auto fdv2 = std::make_unique(); auto fdv1 = std::make_unique(); auto* fdv1_ptr = fdv1.get(); - std::vector> factories; + std::vector> factories; factories.push_back(std::move(fdv2)); factories.push_back(std::move(fdv1)); SourceManager mgr(std::move(factories)); @@ -260,7 +253,7 @@ TEST(SourceManagerTest, SwitchBackToFDv2UnblocksFDv2AndBlocksFDv1) { auto fdv2 = std::make_unique(); auto* fdv2_ptr = fdv2.get(); auto fdv1 = std::make_unique(); - std::vector> factories; + std::vector> factories; factories.push_back(std::move(fdv2)); factories.push_back(std::move(fdv1)); SourceManager mgr(std::move(factories)); @@ -279,7 +272,7 @@ TEST(SourceManagerTest, SwitchBackToFDv2UnblocksFDv2AndBlocksFDv1) { TEST(SourceManagerTest, SwitchBackToFDv2UnblocksTerminallyFailedFDv2Factory) { auto fdv2 = std::make_unique(); auto* fdv2_ptr = fdv2.get(); - std::vector> factories; + std::vector> factories; factories.push_back(std::move(fdv2)); SourceManager mgr(std::move(factories)); diff --git a/libs/server-sdk/src/CMakeLists.txt b/libs/server-sdk/src/CMakeLists.txt index 6be375da6..ff2d7d75d 100644 --- a/libs/server-sdk/src/CMakeLists.txt +++ b/libs/server-sdk/src/CMakeLists.txt @@ -74,10 +74,6 @@ target_sources(${LIBNAME} data_systems/fdv2/polling_synchronizer.cpp data_systems/fdv2/streaming_synchronizer.hpp data_systems/fdv2/streaming_synchronizer.cpp - data_systems/fdv2/conditions.hpp - data_systems/fdv2/conditions.cpp - data_systems/fdv2/source_manager.hpp - data_systems/fdv2/source_manager.cpp data_systems/fdv2/fdv2_data_system.hpp data_systems/fdv2/fdv2_data_system.cpp data_systems/fdv2/fdv1_adapter_synchronizer.hpp diff --git a/libs/server-sdk/src/client_impl.cpp b/libs/server-sdk/src/client_impl.cpp index ba04dda7a..098921ad1 100644 --- a/libs/server-sdk/src/client_impl.cpp +++ b/libs/server-sdk/src/client_impl.cpp @@ -2,7 +2,6 @@ #include "all_flags_state/all_flags_state_builder.hpp" #include "data_systems/background_sync/background_sync_system.hpp" -#include "data_systems/fdv2/conditions.hpp" #include "data_systems/fdv2/fdv2_data_system.hpp" #include "data_systems/fdv2/initializer_factories.hpp" #include "data_systems/fdv2/synchronizer_factories.hpp" diff --git a/libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp b/libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp index 354fb8850..e5b3a52e0 100644 --- a/libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp +++ b/libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp @@ -21,6 +21,18 @@ struct overloaded : Ts... { template overloaded(Ts...) -> overloaded; +// Reduces a source result to the signal the conditions act on. +SourceSignal ClassifyResult(data_interfaces::FDv2SourceResult const& result) { + using Result = data_interfaces::FDv2SourceResult; + if (std::get_if(&result.value)) { + return SourceSignal::kChangeSet; + } + if (std::get_if(&result.value)) { + return SourceSignal::kInterrupted; + } + return SourceSignal::kOther; +} + } // namespace FDv2DataSystem::FDv2DataSystem( @@ -28,10 +40,8 @@ FDv2DataSystem::FDv2DataSystem( initializer_factories, std::vector> synchronizer_factories, - std::unique_ptr - fallback_condition_factory, - std::unique_ptr - recovery_condition_factory, + std::unique_ptr fallback_condition_factory, + std::unique_ptr recovery_condition_factory, boost::asio::any_io_executor ioc, data_components::DataSourceStatusManager* status_manager, Logger const& logger) @@ -293,9 +303,8 @@ void FDv2DataSystem::RunSynchronizerNext() { }); } -void FDv2DataSystem::OnConditionFired( - data_interfaces::IFDv2Condition::Type type) { - using Type = data_interfaces::IFDv2Condition::Type; +void FDv2DataSystem::OnConditionFired(IFDv2Condition::Type type) { + using Type = IFDv2Condition::Type; if (type == Type::kCancelled) { return; } @@ -320,7 +329,7 @@ void FDv2DataSystem::OnConditionFired( } std::unique_ptr FDv2DataSystem::BuildActiveConditions() const { - std::vector> conditions; + std::vector> conditions; // With only one synchronizer available there's nothing to fall back to // or recover from, so leave the conditions empty. if (source_manager_.AvailableSynchronizerCount() == 1) { @@ -346,7 +355,7 @@ void FDv2DataSystem::OnSynchronizerResult( return; } if (active_conditions_) { - active_conditions_->Inform(result); + active_conditions_->Inform(ClassifyResult(result)); } } diff --git a/libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.hpp b/libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.hpp index 3c7147ec1..d3cc61033 100644 --- a/libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.hpp +++ b/libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.hpp @@ -3,15 +3,14 @@ #include "../../data_components/change_notifier/change_notifier.hpp" #include "../../data_components/memory_store/memory_store.hpp" #include "../../data_components/status_notifications/data_source_status_manager.hpp" -#include "../../data_interfaces/source/ifdv2_condition.hpp" #include "../../data_interfaces/source/ifdv2_initializer_factory.hpp" #include "../../data_interfaces/source/ifdv2_synchronizer_factory.hpp" #include "../../data_interfaces/system/idata_system.hpp" -#include "conditions.hpp" -#include "source_manager.hpp" #include #include +#include +#include #include #include @@ -25,6 +24,17 @@ namespace launchdarkly::server_side::data_systems { +// The orchestration primitives the client and server SDKs share. +using internal::data_sources::Conditions; +using internal::data_sources::FallbackConditionFactory; +using internal::data_sources::IFDv2Condition; +using internal::data_sources::IFDv2ConditionFactory; +using internal::data_sources::RecoveryConditionFactory; +using internal::data_sources::SourceSignal; + +using SourceManager = internal::data_sources::SourceManager< + data_interfaces::IFDv2SynchronizerFactory>; + /** * FDv2DataSystem is the IDataSystem implementation for the FDv2 protocol. * It runs a sequence of initializers to populate an in-memory store, then @@ -163,10 +173,8 @@ class FDv2DataSystem final : public data_interfaces::IDataSystem { initializer_factories, std::vector> synchronizer_factories, - std::unique_ptr - fallback_condition_factory, - std::unique_ptr - recovery_condition_factory, + std::unique_ptr fallback_condition_factory, + std::unique_ptr recovery_condition_factory, boost::asio::any_io_executor ioc, data_components::DataSourceStatusManager* status_manager, Logger const& logger); @@ -247,7 +255,7 @@ class FDv2DataSystem final : public data_interfaces::IDataSystem { void StartSynchronizers(); void RunSynchronizerNext(); void OnSynchronizerResult(data_interfaces::FDv2SourceResult result); - void OnConditionFired(data_interfaces::IFDv2Condition::Type type); + void OnConditionFired(IFDv2Condition::Type type); // Schedules an FDv2 recovery attempt after the given TTL. Called with // mutex_ held. TTL of 0 disables the recovery and is a no-op. @@ -273,10 +281,8 @@ class FDv2DataSystem final : public data_interfaces::IDataSystem { boost::asio::any_io_executor const ioc_; std::vector> const initializer_factories_; - std::unique_ptr const - fallback_condition_factory_; - std::unique_ptr const - recovery_condition_factory_; + std::unique_ptr const fallback_condition_factory_; + std::unique_ptr const recovery_condition_factory_; // Non-owning. Lifetime guaranteed by the caller (see constructor doc). data_components::DataSourceStatusManager* const status_manager_; diff --git a/libs/server-sdk/src/data_systems/fdv2/source_manager.cpp b/libs/server-sdk/src/data_systems/fdv2/source_manager.cpp deleted file mode 100644 index 3f020e404..000000000 --- a/libs/server-sdk/src/data_systems/fdv2/source_manager.cpp +++ /dev/null @@ -1,93 +0,0 @@ -#include "source_manager.hpp" - -#include - -namespace launchdarkly::server_side::data_systems { - -using data_interfaces::IFDv2Synchronizer; -using data_interfaces::IFDv2SynchronizerFactory; - -SourceManager::SourceManager( - std::vector> factories) { - synchronizers_.reserve(factories.size()); - for (auto& factory : factories) { - bool const is_fdv1_fallback = factory->IsFDv1Fallback(); - synchronizers_.push_back(SynchronizerFactoryWithState{ - std::move(factory), - is_fdv1_fallback ? State::kBlocked : State::kAvailable, - is_fdv1_fallback}); - } -} - -std::unique_ptr SourceManager::NextSynchronizer() { - if (synchronizers_.empty()) { - current_factory_index_ = -1; - return nullptr; - } - for (std::size_t visited = 0; visited < synchronizers_.size(); ++visited) { - synchronizer_index_ = - (synchronizer_index_ + 1) % static_cast(synchronizers_.size()); - if (synchronizers_[synchronizer_index_].state == State::kAvailable) { - current_factory_index_ = synchronizer_index_; - return synchronizers_[synchronizer_index_].factory->Build(); - } - } - current_factory_index_ = -1; - return nullptr; -} - -void SourceManager::BlockCurrentSynchronizer() { - if (current_factory_index_ >= 0) { - synchronizers_[current_factory_index_].state = State::kBlocked; - } -} - -void SourceManager::ResetSourceIndex() { - synchronizer_index_ = -1; -} - -void SourceManager::SwitchToFDv1Fallback() { - for (auto& entry : synchronizers_) { - entry.state = - entry.is_fdv1_fallback ? State::kAvailable : State::kBlocked; - } - synchronizer_index_ = -1; -} - -void SourceManager::SwitchBackToFDv2() { - for (auto& entry : synchronizers_) { - entry.state = - entry.is_fdv1_fallback ? State::kBlocked : State::kAvailable; - } - synchronizer_index_ = -1; -} - -bool SourceManager::IsPrimeSynchronizer() const { - for (std::size_t i = 0; i < synchronizers_.size(); ++i) { - if (synchronizers_[i].state == State::kAvailable) { - return synchronizer_index_ == static_cast(i); - } - } - return false; -} - -std::size_t SourceManager::AvailableSynchronizerCount() const { - std::size_t count = 0; - for (auto const& s : synchronizers_) { - if (s.state == State::kAvailable) { - ++count; - } - } - return count; -} - -std::size_t SourceManager::SynchronizerCount() const { - return synchronizers_.size(); -} - -bool SourceManager::IsCurrentSynchronizerFDv1Fallback() const { - return current_factory_index_ >= 0 && - synchronizers_[current_factory_index_].is_fdv1_fallback; -} - -} // namespace launchdarkly::server_side::data_systems