feat: cyclic synchronizer rotation with block/unblock state

beekld · beekld · commit 810f0d3e1caf · 2026-05-12T13:25:53.000-07:00
diff --git a/libs/server-sdk/src/CMakeLists.txt b/libs/server-sdk/src/CMakeLists.txt
@@ -62,6 +62,8 @@ target_sources(${LIBNAME}
         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/background_sync/sources/streaming/streaming_data_source.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
@@ -37,7 +37,6 @@ FDv2DataSystem::FDv2DataSystem(
     : logger_(logger),
       ioc_(std::move(ioc)),
       initializer_factories_(std::move(initializer_factories)),
-      synchronizer_factories_(std::move(synchronizer_factories)),
       fallback_condition_factory_(std::move(fallback_condition_factory)),
       recovery_condition_factory_(std::move(recovery_condition_factory)),
       status_manager_(status_manager),
@@ -47,7 +46,7 @@ FDv2DataSystem::FDv2DataSystem(
       closed_(false),
       selector_(),
       initializer_index_(0),
-      synchronizer_index_(0),
+      source_manager_(std::move(synchronizer_factories)),
       active_initializer_(nullptr),
       active_synchronizer_(nullptr),
       active_conditions_(nullptr) {}
@@ -101,7 +100,8 @@ void FDv2DataSystem::Initialize() {
     assert(!already_called && "Initialize() must be called at most once");
 
     LD_LOG(logger_, LogLevel::kInfo) << Identity() << ": starting";
-    if (initializer_factories_.empty() && synchronizer_factories_.empty()) {
+    if (initializer_factories_.empty() &&
+        source_manager_.SynchronizerCount() == 0) {
         // Offline mode: empty store is the canonical state.
         status_manager_->SetState(DataSourceStatus::DataSourceState::kValid);
         return;
@@ -196,30 +196,27 @@ void FDv2DataSystem::OnInitializerResult(
 
 void FDv2DataSystem::StartSynchronizers() {
     bool exhausted = false;
-    bool cycled_synchronizers = false;
+    // True if at least one synchronizer factory was configured at construction.
+    bool any_synchronizers_configured = false;
     {
         std::lock_guard<std::mutex> lock(mutex_);
         if (closed_) {
             return;
         }
-        if (synchronizer_index_ >= synchronizer_factories_.size()) {
-            exhausted = true;
-            cycled_synchronizers = synchronizer_index_ > 0;
+        active_synchronizer_ = source_manager_.NextSynchronizer();
+        if (active_synchronizer_) {
+            active_conditions_ = BuildActiveConditions();
         } else {
-            auto& factory = synchronizer_factories_[synchronizer_index_];
-            active_synchronizer_ = factory->Build();
-            active_conditions_ =
-                BuildConditionsForSynchronizer(synchronizer_index_);
-            ++synchronizer_index_;
+            exhausted = true;
+            any_synchronizers_configured =
+                source_manager_.SynchronizerCount() > 0;
         }
     }
 
     if (exhausted) {
-        // kOff when we can't continue updating; init-only with data stays
-        // kValid.
-        if (cycled_synchronizers || !store_.Initialized()) {
+        if (any_synchronizers_configured || !store_.Initialized()) {
             std::string const message =
-                cycled_synchronizers
+                any_synchronizers_configured
                     ? "all data source acquisition methods have been exhausted"
                     : "all initializers exhausted and no synchronizers "
                       "configured";
@@ -274,7 +271,7 @@ void FDv2DataSystem::OnConditionFired(
         if (type == Type::kRecovery) {
             LD_LOG(logger_, LogLevel::kInfo)
                 << Identity() << ": recovery condition met";
-            synchronizer_index_ = 0;
+            source_manager_.ResetSourceIndex();
         } else {
             LD_LOG(logger_, LogLevel::kInfo)
                 << Identity() << ": fallback condition met";
@@ -283,16 +280,18 @@ void FDv2DataSystem::OnConditionFired(
     StartSynchronizers();
 }
 
-std::unique_ptr<Conditions> FDv2DataSystem::BuildConditionsForSynchronizer(
-    std::size_t synchronizer_position) const {
+std::unique_ptr<Conditions> FDv2DataSystem::BuildActiveConditions() const {
     std::vector<std::unique_ptr<data_interfaces::IFDv2Condition>> conditions;
-    if (synchronizer_factories_.size() <= 1) {
+    // 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<Conditions>(std::move(conditions));
     }
     if (fallback_condition_factory_) {
         conditions.push_back(fallback_condition_factory_->Build());
     }
-    if (synchronizer_position > 0 && recovery_condition_factory_) {
+    // 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<Conditions>(std::move(conditions));
@@ -356,6 +355,7 @@ void FDv2DataSystem::OnSynchronizerResult(
             return;
         }
         if (advance) {
+            source_manager_.BlockCurrentSynchronizer();
             active_synchronizer_.reset();
             active_conditions_.reset();
         }
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
@@ -8,6 +8,7 @@
 #include "../../data_interfaces/source/ifdv2_synchronizer_factory.hpp"
 #include "../../data_interfaces/system/idata_system.hpp"
 #include "conditions.hpp"
+#include "source_manager.hpp"
 
 #include <launchdarkly/data_model/selector.hpp>
 #include <launchdarkly/logging/logger.hpp>
@@ -104,16 +105,18 @@ namespace launchdarkly::server_side::data_systems {
  *               |
  *               | (N exhausted, or basis received)
  *               v
- *     +-------------------+    synchronizer #M's Next returns:
+ *     +-------------------+    active synchronizer's Next returns:
  *     |  Synchronizer     |      ChangeSet      -> apply, loop
  *     |  phase            |      Interrupted    -> loop (source self-retries)
- *     |  M = 0, 1, 2, ... |      Timeout        -> loop
- *     |                   |      Goodbye        -> loop (source self-restarts)
- *     |                   |      TerminalError  -> M += 1
- *     |                   |      Shutdown       -> [Closed]
+ *     |  (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
+ *                   +---+
  *               |
- *               | (M exhausted)
+ *               | (all synchronizers blocked)
  *               v
  *     [Done; final status preserved]
  *
@@ -248,11 +251,9 @@ class FDv2DataSystem final : public data_interfaces::IDataSystem {
     void OnSynchronizerResult(data_interfaces::FDv2SourceResult result);
     void OnConditionFired(data_interfaces::IFDv2Condition::Type type);
 
-    // Builds the conditions to apply to a synchronizer at the given chain
-    // position. Reads only const-after-construction state, so no
-    // synchronization is required.
-    std::unique_ptr<Conditions> BuildConditionsForSynchronizer(
-        std::size_t synchronizer_position) const;
+    // Builds the conditions to apply to the currently active synchronizer.
+    // Must be called with mutex_ held; reads source_manager_ state.
+    std::unique_ptr<Conditions> BuildActiveConditions() const;
 
     // Applies a typed FDv2 changeset to the in-memory store and updates the
     // tracked selector if the changeset's selector is non-empty.
@@ -266,9 +267,6 @@ class FDv2DataSystem final : public data_interfaces::IDataSystem {
     boost::asio::any_io_executor const ioc_;
     std::vector<std::unique_ptr<data_interfaces::IFDv2InitializerFactory>> const
         initializer_factories_;
-    std::vector<
-        std::unique_ptr<data_interfaces::IFDv2SynchronizerFactory>> const
-        synchronizer_factories_;
     std::unique_ptr<data_interfaces::IFDv2ConditionFactory> const
         fallback_condition_factory_;
     std::unique_ptr<data_interfaces::IFDv2ConditionFactory> const
@@ -290,7 +288,7 @@ class FDv2DataSystem final : public data_interfaces::IDataSystem {
     bool closed_;
     data_model::Selector selector_;
     std::size_t initializer_index_;
-    std::size_t synchronizer_index_;
+    SourceManager source_manager_;
     std::unique_ptr<data_interfaces::IFDv2Initializer> active_initializer_;
     std::unique_ptr<data_interfaces::IFDv2Synchronizer> active_synchronizer_;
     std::unique_ptr<Conditions> active_conditions_;
diff --git a/libs/server-sdk/src/data_systems/fdv2/source_manager.cpp b/libs/server-sdk/src/data_systems/fdv2/source_manager.cpp
@@ -0,0 +1,75 @@
+#include "source_manager.hpp"
+
+#include <utility>
+
+namespace launchdarkly::server_side::data_systems {
+
+using data_interfaces::IFDv2Synchronizer;
+using data_interfaces::IFDv2SynchronizerFactory;
+
+SourceManager::SourceManager(
+    std::vector<std::unique_ptr<IFDv2SynchronizerFactory>> factories) {
+    synchronizers_.reserve(factories.size());
+    for (auto& factory : factories) {
+        synchronizers_.push_back(
+            SynchronizerFactoryWithState{std::move(factory), State::kAvailable,
+                                         /*is_fdv1_fallback=*/false});
+    }
+}
+
+std::unique_ptr<IFDv2Synchronizer> 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<int>(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;
+}
+
+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<int>(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
diff --git a/libs/server-sdk/src/data_systems/fdv2/source_manager.hpp b/libs/server-sdk/src/data_systems/fdv2/source_manager.hpp
@@ -0,0 +1,104 @@
+#pragma once
+
+#include "../../data_interfaces/source/ifdv2_synchronizer.hpp"
+#include "../../data_interfaces/source/ifdv2_synchronizer_factory.hpp"
+
+#include <cstddef>
+#include <memory>
+#include <vector>
+
+namespace launchdarkly::server_side::data_systems {
+
+/**
+ * Manages a list of synchronizer factories together with per-factory state
+ * (Available / Blocked) and the index of the currently active factory.
+ *
+ * Iteration is cyclic: NextSynchronizer advances past the end and wraps to
+ * the beginning, skipping any factory in the Blocked state. A factory enters
+ * the Blocked state when BlockCurrentSynchronizer is called (typically on a
+ * terminal error). Once blocked, a factory is not eligible to be built again
+ * until unblocked.
+ *
+ * ResetSourceIndex causes the next call to start iteration at index 0 — used
+ * 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.
+ *
+ * Not thread-safe. The caller is responsible for serializing all calls.
+ */
+class SourceManager {
+   public:
+    explicit SourceManager(
+        std::vector<std::unique_ptr<data_interfaces::IFDv2SynchronizerFactory>>
+            factories);
+
+    /**
+     * Advances to the next Available synchronizer factory (wrapping past the
+     * end), builds a synchronizer instance from it, and records that factory
+     * as the current one for subsequent queries. Returns nullptr if no
+     * Available factory exists.
+     */
+    std::unique_ptr<data_interfaces::IFDv2Synchronizer> NextSynchronizer();
+
+    /**
+     * Marks the currently tracked factory as Blocked. No-op if no factory is
+     * currently tracked.
+     */
+    void BlockCurrentSynchronizer();
+
+    /**
+     * Resets the iteration cursor so that the next call to NextSynchronizer
+     * begins searching from index 0.
+     */
+    void ResetSourceIndex();
+
+    /**
+     * 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;
+
+    /**
+     * Returns the count of factories not in the Blocked state.
+     */
+    [[nodiscard]] std::size_t AvailableSynchronizerCount() const;
+
+    /**
+     * 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;
+
+    /**
+     * Returns true if the currently tracked factory was configured as the
+     * FDv1 fallback synchronizer. Always false until the FDv1 fallback
+     * directive is implemented.
+     */
+    [[nodiscard]] bool IsCurrentSynchronizerFDv1Fallback() const;
+
+    SourceManager(SourceManager const&) = delete;
+    SourceManager(SourceManager&&) = delete;
+    SourceManager& operator=(SourceManager const&) = delete;
+    SourceManager& operator=(SourceManager&&) = delete;
+    ~SourceManager() = default;
+
+   private:
+    enum class State { kAvailable, kBlocked };
+
+    struct SynchronizerFactoryWithState {
+        std::unique_ptr<data_interfaces::IFDv2SynchronizerFactory> factory;
+        State state = State::kAvailable;
+        bool is_fdv1_fallback = false;
+    };
+
+    std::vector<SynchronizerFactoryWithState> synchronizers_;
+    // Iteration cursor; -1 means "start at 0 on next call".
+    int synchronizer_index_ = -1;
+    // Index of the most recently returned factory; -1 if none.
+    int current_factory_index_ = -1;
+};
+
+}  // namespace launchdarkly::server_side::data_systems
diff --git a/libs/server-sdk/tests/fdv2_data_system_test.cpp b/libs/server-sdk/tests/fdv2_data_system_test.cpp
diff --git a/libs/server-sdk/tests/source_manager_test.cpp b/libs/server-sdk/tests/source_manager_test.cpp
