diff --git a/libs/server-sdk/src/data_interfaces/source/ifdv2_synchronizer_factory.hpp b/libs/server-sdk/src/data_interfaces/source/ifdv2_synchronizer_factory.hpp index fc3c6421a..ba130e768 100644 --- a/libs/server-sdk/src/data_interfaces/source/ifdv2_synchronizer_factory.hpp +++ b/libs/server-sdk/src/data_interfaces/source/ifdv2_synchronizer_factory.hpp @@ -14,6 +14,8 @@ class IFDv2SynchronizerFactory { public: virtual std::unique_ptr Build() = 0; + [[nodiscard]] virtual bool IsFDv1Fallback() const { return false; } + virtual ~IFDv2SynchronizerFactory() = default; IFDv2SynchronizerFactory(IFDv2SynchronizerFactory const&) = delete; IFDv2SynchronizerFactory(IFDv2SynchronizerFactory&&) = delete; 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 b3caa5aba..ba66bfe4f 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 @@ -185,6 +185,12 @@ void FDv2DataSystem::OnInitializerResult( if (closed_ || got_shutdown) { return; } + if (result.fdv1_fallback) { + LD_LOG(logger_, LogLevel::kInfo) + << Identity() << ": FDv1 fallback engaged"; + source_manager_.SwitchToFDv1Fallback(); + got_basis = true; + } } if (got_basis) { @@ -349,7 +355,14 @@ void FDv2DataSystem::OnSynchronizerResult( active_conditions_.reset(); return; } - if (advance) { + if (result.fdv1_fallback) { + LD_LOG(logger_, LogLevel::kInfo) + << Identity() << ": FDv1 fallback engaged"; + source_manager_.SwitchToFDv1Fallback(); + active_synchronizer_.reset(); + active_conditions_.reset(); + advance = true; + } else if (advance) { source_manager_.BlockCurrentSynchronizer(); active_synchronizer_.reset(); active_conditions_.reset(); diff --git a/libs/server-sdk/src/data_systems/fdv2/source_manager.cpp b/libs/server-sdk/src/data_systems/fdv2/source_manager.cpp index 362f13efd..4af2405a6 100644 --- a/libs/server-sdk/src/data_systems/fdv2/source_manager.cpp +++ b/libs/server-sdk/src/data_systems/fdv2/source_manager.cpp @@ -11,9 +11,11 @@ SourceManager::SourceManager( std::vector> factories) { synchronizers_.reserve(factories.size()); for (auto& factory : factories) { - synchronizers_.push_back( - SynchronizerFactoryWithState{std::move(factory), State::kAvailable, - /*is_fdv1_fallback=*/false}); + bool const is_fdv1_fallback = factory->IsFDv1Fallback(); + synchronizers_.push_back(SynchronizerFactoryWithState{ + std::move(factory), + is_fdv1_fallback ? State::kBlocked : State::kAvailable, + is_fdv1_fallback}); } } @@ -44,6 +46,14 @@ 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; +} + bool SourceManager::IsPrimeSynchronizer() const { for (std::size_t i = 0; i < synchronizers_.size(); ++i) { if (synchronizers_[i].state == State::kAvailable) { diff --git a/libs/server-sdk/src/data_systems/fdv2/source_manager.hpp b/libs/server-sdk/src/data_systems/fdv2/source_manager.hpp index 1f310bfed..bd2347ccf 100644 --- a/libs/server-sdk/src/data_systems/fdv2/source_manager.hpp +++ b/libs/server-sdk/src/data_systems/fdv2/source_manager.hpp @@ -23,8 +23,7 @@ namespace launchdarkly::server_side::data_systems { * by recovery, which wants to fall back to the most-preferred Available * synchronizer. * - * Each factory also carries an is_fdv1_fallback flag, currently always - * false. TODO: populate when the FDv1 fallback directive is implemented. + * Factories whose IsFDv1Fallback() returns true start in the Blocked state. * * Not thread-safe. The caller is responsible for serializing all calls. */ @@ -54,6 +53,14 @@ class SourceManager { */ void ResetSourceIndex(); + /** + * Blocks every non-FDv1 factory and unblocks the FDv1 fallback factory, + * if one was configured. Resets the iteration cursor so the next call to + * NextSynchronizer returns the FDv1 fallback. If no FDv1 fallback factory + * was configured, every factory is left blocked. + */ + void SwitchToFDv1Fallback(); + /** * Returns true if the currently tracked factory is the first Available * factory in the list. Returns false if no factory is currently tracked. @@ -73,9 +80,8 @@ class SourceManager { [[nodiscard]] std::size_t SynchronizerCount() const; /** - * Returns true if the currently tracked factory was configured as the - * FDv1 fallback synchronizer. Always false until the FDv1 fallback - * directive is implemented. + * Returns true if the currently tracked factory is the FDv1 fallback + * synchronizer. */ [[nodiscard]] bool IsCurrentSynchronizerFDv1Fallback() const; diff --git a/libs/server-sdk/tests/fdv2_data_system_test.cpp b/libs/server-sdk/tests/fdv2_data_system_test.cpp index 48d9b49d1..72ef65ff4 100644 --- a/libs/server-sdk/tests/fdv2_data_system_test.cpp +++ b/libs/server-sdk/tests/fdv2_data_system_test.cpp @@ -145,6 +145,15 @@ class OneShotSynchronizerFactory : public IFDv2SynchronizerFactory { std::unique_ptr source_; }; +class FDv1FallbackOneShotFactory : public OneShotSynchronizerFactory { + public: + explicit FDv1FallbackOneShotFactory( + std::unique_ptr source) + : OneShotSynchronizerFactory(std::move(source)) {} + + bool IsFDv1Fallback() const override { return true; } +}; + // Returns each pre-supplied source in order on successive Build() calls. // Returns nullptr once the supply is exhausted. Used in tests that exercise // wrap-around or recovery, where the same factory is built more than once. @@ -1090,6 +1099,129 @@ TEST(FDv2DataSystemTest, SingleSynchronizerHasNoFallbackArmed) { status_manager.Status().State()); } +// ============================================================================ +// FDv1 fallback directive +// ============================================================================ + +TEST(FDv2DataSystemTest, SynchronizerFdv1FlagSwitchesToFdv1Adapter) { + auto logger = MakeNullLogger(); + boost::asio::io_context ioc; + data_components::DataSourceStatusManager status_manager; + + // FDv2 synchronizer emits a ChangeSet with the directive, then closes. + auto fdv2_sync = + std::make_unique(std::vector{[]() { + FDv2SourceResult r{FDv2SourceResult::ChangeSet{ + data_model::ChangeSet{ + data_model::ChangeSetType::kNone, + {}, + data_model::Selector{}}}}; + r.fdv1_fallback = true; + return r; + }()}); + auto fdv2_factory = + std::make_unique(std::move(fdv2_sync)); + + // FDv1 adapter returns Shutdown when reached, ending orchestration. + auto fdv1_sync = + std::make_unique(std::vector{}); + auto fdv1_factory = + std::make_unique(std::move(fdv1_sync)); + auto* fdv1_factory_ptr = fdv1_factory.get(); + + std::vector> synchronizers; + synchronizers.push_back(std::move(fdv2_factory)); + synchronizers.push_back(std::move(fdv1_factory)); + + FDv2DataSystem ds({}, std::move(synchronizers), + /*fallback_condition_factory=*/nullptr, + /*recovery_condition_factory=*/nullptr, + ioc.get_executor(), &status_manager, logger); + ds.Initialize(); + ioc.run(); + + EXPECT_EQ(1, fdv1_factory_ptr->build_count_); +} + +TEST(FDv2DataSystemTest, SynchronizerFdv1FlagWithoutAdapterTransitionsOff) { + auto logger = MakeNullLogger(); + boost::asio::io_context ioc; + data_components::DataSourceStatusManager status_manager; + + auto fdv2_sync = + std::make_unique(std::vector{[]() { + FDv2SourceResult r{ + FDv2SourceResult::Interrupted{FDv2SourceResult::ErrorInfo{ + FDv2SourceResult::ErrorInfo::ErrorKind::kErrorResponse, + /*status_code=*/418, "directive", + std::chrono::system_clock::now()}}}; + r.fdv1_fallback = true; + return r; + }()}); + auto fdv2_factory = + std::make_unique(std::move(fdv2_sync)); + + std::vector> synchronizers; + synchronizers.push_back(std::move(fdv2_factory)); + + FDv2DataSystem ds({}, std::move(synchronizers), + /*fallback_condition_factory=*/nullptr, + /*recovery_condition_factory=*/nullptr, + ioc.get_executor(), &status_manager, logger); + ds.Initialize(); + ioc.run(); + + EXPECT_EQ(DataSourceStatus::DataSourceState::kOff, + status_manager.Status().State()); +} + +TEST(FDv2DataSystemTest, InitializerFdv1FlagSwitchesToFdv1Adapter) { + auto logger = MakeNullLogger(); + boost::asio::io_context ioc; + data_components::DataSourceStatusManager status_manager; + + // Initializer returns Interrupted with the directive set. + FDv2SourceResult init_result{ + FDv2SourceResult::Interrupted{FDv2SourceResult::ErrorInfo{ + FDv2SourceResult::ErrorInfo::ErrorKind::kErrorResponse, + /*status_code=*/418, "directive", + std::chrono::system_clock::now()}}}; + init_result.fdv1_fallback = true; + auto initializer = + std::make_unique(std::move(init_result)); + + std::vector> initializers; + initializers.push_back( + std::make_unique(std::move(initializer))); + + auto fdv2_sync = + std::make_unique(std::vector{}); + auto fdv2_factory = + std::make_unique(std::move(fdv2_sync)); + auto* fdv2_factory_ptr = fdv2_factory.get(); + + auto fdv1_sync = + std::make_unique(std::vector{}); + auto fdv1_factory = + std::make_unique(std::move(fdv1_sync)); + auto* fdv1_factory_ptr = fdv1_factory.get(); + + std::vector> synchronizers; + synchronizers.push_back(std::move(fdv2_factory)); + synchronizers.push_back(std::move(fdv1_factory)); + + FDv2DataSystem ds(std::move(initializers), std::move(synchronizers), + /*fallback_condition_factory=*/nullptr, + /*recovery_condition_factory=*/nullptr, + ioc.get_executor(), &status_manager, logger); + ds.Initialize(); + ioc.run(); + + // FDv2 synchronizer was skipped; FDv1 adapter was built and ran. + EXPECT_EQ(0, fdv2_factory_ptr->build_count_); + EXPECT_EQ(1, fdv1_factory_ptr->build_count_); +} + // ============================================================================ // Destruction protocol: in-flight orchestration // ============================================================================ diff --git a/libs/server-sdk/tests/source_manager_test.cpp b/libs/server-sdk/tests/source_manager_test.cpp index 244ca6400..77b207b46 100644 --- a/libs/server-sdk/tests/source_manager_test.cpp +++ b/libs/server-sdk/tests/source_manager_test.cpp @@ -43,6 +43,11 @@ class CountingFactory : public IFDv2SynchronizerFactory { int build_count = 0; }; +class FDv1FallbackFactory : public CountingFactory { + public: + bool IsFDv1Fallback() const override { return true; } +}; + } // namespace TEST(SourceManagerTest, EmptyManagerReportsZeroAvailable) { @@ -176,7 +181,7 @@ TEST(SourceManagerTest, ResetSourceIndexSkipsBlockedFirstFactory) { EXPECT_EQ(1, f1_ptr->build_count); } -TEST(SourceManagerTest, IsCurrentSynchronizerFDv1FallbackAlwaysFalse) { +TEST(SourceManagerTest, IsCurrentSynchronizerFDv1FallbackFalseForFDv2Factory) { auto f0 = std::make_unique(); std::vector> factories; factories.push_back(std::move(f0)); @@ -185,3 +190,68 @@ TEST(SourceManagerTest, IsCurrentSynchronizerFDv1FallbackAlwaysFalse) { mgr.NextSynchronizer(); EXPECT_FALSE(mgr.IsCurrentSynchronizerFDv1Fallback()); } + +TEST(SourceManagerTest, FDv1FallbackFactoryStartsBlockedAndIsSkipped) { + auto fdv2 = std::make_unique(); + auto fdv1 = std::make_unique(); + auto* fdv1_ptr = fdv1.get(); + std::vector> factories; + factories.push_back(std::move(fdv2)); + factories.push_back(std::move(fdv1)); + SourceManager mgr(std::move(factories)); + + EXPECT_EQ(1u, mgr.AvailableSynchronizerCount()); + mgr.NextSynchronizer(); + EXPECT_FALSE(mgr.IsCurrentSynchronizerFDv1Fallback()); + EXPECT_EQ(0, fdv1_ptr->build_count); +} + +TEST(SourceManagerTest, SwitchToFDv1FallbackBlocksFDv2AndUnblocksFDv1) { + auto fdv2 = std::make_unique(); + auto fdv1 = std::make_unique(); + auto* fdv1_ptr = fdv1.get(); + std::vector> factories; + factories.push_back(std::move(fdv2)); + factories.push_back(std::move(fdv1)); + SourceManager mgr(std::move(factories)); + + mgr.SwitchToFDv1Fallback(); + + EXPECT_EQ(1u, mgr.AvailableSynchronizerCount()); + auto sync = mgr.NextSynchronizer(); + ASSERT_NE(sync, nullptr); + EXPECT_EQ(1, fdv1_ptr->build_count); + EXPECT_TRUE(mgr.IsCurrentSynchronizerFDv1Fallback()); +} + +TEST(SourceManagerTest, SwitchToFDv1FallbackWithoutAdapterBlocksEverything) { + auto fdv2 = std::make_unique(); + std::vector> factories; + factories.push_back(std::move(fdv2)); + SourceManager mgr(std::move(factories)); + + mgr.SwitchToFDv1Fallback(); + + EXPECT_EQ(0u, mgr.AvailableSynchronizerCount()); + EXPECT_EQ(nullptr, mgr.NextSynchronizer()); +} + +TEST(SourceManagerTest, SwitchToFDv1FallbackUnblocksPreviouslyBlockedFDv2) { + auto fdv2 = std::make_unique(); + auto fdv1 = std::make_unique(); + auto* fdv1_ptr = fdv1.get(); + std::vector> factories; + factories.push_back(std::move(fdv2)); + factories.push_back(std::move(fdv1)); + SourceManager mgr(std::move(factories)); + + mgr.NextSynchronizer(); + mgr.BlockCurrentSynchronizer(); + mgr.SwitchToFDv1Fallback(); + + EXPECT_EQ(1u, mgr.AvailableSynchronizerCount()); + auto sync = mgr.NextSynchronizer(); + ASSERT_NE(sync, nullptr); + EXPECT_EQ(1, fdv1_ptr->build_count); + EXPECT_TRUE(mgr.IsCurrentSynchronizerFDv1Fallback()); +}