fix: server FDv2 goodbye behavior and orchestration polish

beekld · beekld · commit d154cbe14c12 · 2026-05-05T22:07:05.000-07:00
diff --git a/libs/server-sdk/src/data_components/change_notifier/change_notifier.cpp b/libs/server-sdk/src/data_components/change_notifier/change_notifier.cpp
@@ -10,6 +10,7 @@ namespace launchdarkly::server_side::data_components {
 
 namespace {
 
+// Lets std::visit dispatch to a different lambda per variant alternative.
 template <class... Ts>
 struct overloaded : Ts... {
     using Ts::operator()...;
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
@@ -4,6 +4,7 @@
 
 #include <boost/asio/post.hpp>
 
+#include <cassert>
 #include <chrono>
 #include <utility>
 #include <variant>
@@ -12,6 +13,7 @@ namespace launchdarkly::server_side::data_systems {
 
 namespace {
 
+// Lets std::visit dispatch to a different lambda per variant alternative.
 template <class... Ts>
 struct overloaded : Ts... {
     using Ts::operator()...;
@@ -39,6 +41,7 @@ FDv2DataSystem::FDv2DataSystem(
       status_manager_(status_manager),
       store_(),
       change_notifier_(store_, store_),
+      initialize_called_(false),
       closed_(false),
       selector_(),
       initializer_index_(0),
@@ -88,6 +91,9 @@ std::string const& FDv2DataSystem::Identity() const {
 }
 
 void FDv2DataSystem::Initialize() {
+    bool const already_called = initialize_called_.exchange(true);
+    assert(!already_called && "Initialize() must be called at most once");
+
     LD_LOG(logger_, LogLevel::kInfo) << Identity() << ": starting";
     if (initializer_factories_.empty() && synchronizer_factories_.empty()) {
         // Offline mode: empty store is the canonical state.
@@ -98,8 +104,6 @@ void FDv2DataSystem::Initialize() {
 }
 
 void FDv2DataSystem::RunNextInitializer() {
-    auto future = async::MakeFuture(data_interfaces::FDv2SourceResult{
-        data_interfaces::FDv2SourceResult::Shutdown{}});
     bool exhausted = false;
     {
         std::lock_guard<std::mutex> lock(mutex_);
@@ -111,22 +115,21 @@ void FDv2DataSystem::RunNextInitializer() {
         } else {
             auto& factory = initializer_factories_[initializer_index_++];
             active_initializer_ = factory->Build();
-            future = active_initializer_->Run();
+            active_initializer_->Run().Then(
+                [this](data_interfaces::FDv2SourceResult const& result)
+                    -> std::monostate {
+                    OnInitializerResult(result);
+                    return {};
+                },
+                [ioc = ioc_](async::Continuation<void()> work) {
+                    boost::asio::post(ioc, std::move(work));
+                });
         }
     }
 
     if (exhausted) {
         StartSynchronizers();
-        return;
     }
-
-    std::move(future).Then(
-        [this](
-            data_interfaces::FDv2SourceResult const& result) -> std::monostate {
-            OnInitializerResult(result);
-            return {};
-        },
-        async::kInlineExecutor);
 }
 
 void FDv2DataSystem::OnInitializerResult(
@@ -164,13 +167,12 @@ void FDv2DataSystem::OnInitializerResult(
                     te.error.Kind(), te.error.Message());
             },
             [&](Result::Goodbye const&) {
-                LD_LOG(logger_, LogLevel::kWarn)
-                    << Identity()
-                    << ": initializer received unexpected goodbye";
+                LD_LOG(logger_, LogLevel::kDebug)
+                    << Identity() << ": ignoring goodbye from initializer";
             },
             [&](Result::Timeout const&) {
-                LD_LOG(logger_, LogLevel::kWarn)
-                    << Identity() << ": initializer timed out (unexpected)";
+                LD_LOG(logger_, LogLevel::kDebug)
+                    << Identity() << ": ignoring timeout from initializer";
             },
         },
         result.value);
@@ -192,47 +194,56 @@ void FDv2DataSystem::OnInitializerResult(
 
 void FDv2DataSystem::StartSynchronizers() {
     bool exhausted = false;
+    bool cycled_synchronizers = false;
     {
         std::lock_guard<std::mutex> lock(mutex_);
         if (closed_) {
             return;
         }
         if (synchronizer_index_ >= synchronizer_factories_.size()) {
             exhausted = true;
+            cycled_synchronizers = synchronizer_index_ > 0;
         } else {
             auto& factory = synchronizer_factories_[synchronizer_index_++];
             active_synchronizer_ = factory->Build();
         }
     }
 
     if (exhausted) {
-        LD_LOG(logger_, LogLevel::kWarn)
-            << Identity() << ": no synchronizers available";
+        // kOff when we can't continue updating; init-only with data stays
+        // kValid.
+        if (cycled_synchronizers || !store_.Initialized()) {
+            std::string const message =
+                cycled_synchronizers
+                    ? "all data source acquisition methods have been exhausted"
+                    : "all initializers exhausted and no synchronizers "
+                      "configured";
+            LD_LOG(logger_, LogLevel::kWarn) << Identity() << ": " << message;
+            status_manager_->SetState(
+                DataSourceStatus::DataSourceState::kOff,
+                DataSourceStatus::ErrorInfo::ErrorKind::kUnknown, message);
+        }
         return;
     }
 
     RunSynchronizerNext();
 }
 
 void FDv2DataSystem::RunSynchronizerNext() {
-    auto future = async::MakeFuture(data_interfaces::FDv2SourceResult{
-        data_interfaces::FDv2SourceResult::Shutdown{}});
-    {
-        std::lock_guard<std::mutex> lock(mutex_);
-        if (closed_ || !active_synchronizer_) {
-            return;
-        }
-        future =
-            active_synchronizer_->Next(kSynchronizerNextTimeout, selector_);
+    std::lock_guard<std::mutex> lock(mutex_);
+    if (closed_ || !active_synchronizer_) {
+        return;
     }
-
-    std::move(future).Then(
-        [this](
-            data_interfaces::FDv2SourceResult const& result) -> std::monostate {
-            OnSynchronizerResult(result);
-            return {};
-        },
-        async::kInlineExecutor);
+    active_synchronizer_->Next(kSynchronizerNextTimeout, selector_)
+        .Then(
+            [this](data_interfaces::FDv2SourceResult const& result)
+                -> std::monostate {
+                OnSynchronizerResult(result);
+                return {};
+            },
+            [ioc = ioc_](async::Continuation<void()> work) {
+                boost::asio::post(ioc, std::move(work));
+            });
 }
 
 void FDv2DataSystem::OnSynchronizerResult(
@@ -266,10 +277,11 @@ void FDv2DataSystem::OnSynchronizerResult(
                 advance = true;
             },
             [&](Result::Goodbye const& gb) {
-                LD_LOG(logger_, LogLevel::kInfo)
-                    << Identity() << ": synchronizer goodbye"
+                // The synchronizer handles goodbye internally (reconnects);
+                // the orchestrator just loops on the same source.
+                LD_LOG(logger_, LogLevel::kDebug)
+                    << Identity() << ": ignoring goodbye from synchronizer"
                     << (gb.reason ? ": " + *gb.reason : "");
-                advance = true;
             },
             [&](Result::Timeout const&) {
                 LD_LOG(logger_, LogLevel::kDebug)
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
@@ -12,6 +12,7 @@
 
 #include <boost/asio/any_io_executor.hpp>
 
+#include <atomic>
 #include <cstddef>
 #include <memory>
 #include <mutex>
@@ -105,7 +106,8 @@ namespace launchdarkly::server_side::data_systems {
  *     |  Synchronizer     |      ChangeSet      -> apply, loop
  *     |  phase            |      Interrupted    -> loop (source self-retries)
  *     |  M = 0, 1, 2, ... |      Timeout        -> loop
- *     |                   |      Goodbye/Term   -> M += 1
+ *     |                   |      Goodbye        -> loop (source self-restarts)
+ *     |                   |      TerminalError  -> M += 1
  *     |                   |      Shutdown       -> [Closed]
  *     +-------------------+
  *               |
@@ -121,9 +123,13 @@ namespace launchdarkly::server_side::data_systems {
  *                              kInterrupted on Interrupted / TerminalError
  *                              (filtered to kInitializing while still in
  *                              the initializer phase if not yet Valid).
- *   kValid                  -> kInterrupted on errors; kOff in destructor.
+ *                              kOff if all initializers exhaust without data
+ *                              and no synchronizers are configured.
+ *   kValid                  -> kInterrupted on errors; kOff in destructor or
+ *                              when synchronizers cycle through and exhaust.
  *   kInterrupted            -> kValid on next successful ChangeSet apply;
- *                              kOff in destructor.
+ *                              kOff in destructor or on synchronizer
+ *                              exhaustion.
  *   kOff                    -> terminal.
  */
 class FDv2DataSystem final : public data_interfaces::IDataSystem {
@@ -217,8 +223,10 @@ class FDv2DataSystem final : public data_interfaces::IDataSystem {
     void Close();
 
     // Orchestration-loop methods. Each step chains the next via Future::Then,
-    // so at most one is in flight at a time. mutex_ guards shared state
-    // against the destructor's Close() running concurrently with a callback.
+    // so at most one step has a pending continuation at any time. mutex_
+    // provides mutual exclusion for orchestration state when callbacks run on
+    // different executor threads, and lets Close() safely tear down active
+    // sources from any thread.
 
     void RunNextInitializer();
     void OnInitializerResult(data_interfaces::FDv2SourceResult result);
@@ -250,6 +258,9 @@ class FDv2DataSystem final : public data_interfaces::IDataSystem {
     // safe.
     data_components::ChangeNotifier change_notifier_;
 
+    // Set by Initialize() to detect repeat or concurrent calls.
+    std::atomic_bool initialize_called_;
+
     // Orchestration state, guarded by mutex_.
     std::mutex mutex_;
     bool closed_;
diff --git a/libs/server-sdk/src/data_systems/fdv2/streaming_synchronizer.cpp b/libs/server-sdk/src/data_systems/fdv2/streaming_synchronizer.cpp
@@ -210,6 +210,14 @@ void FDv2StreamingSynchronizer::State::OnEvent(sse::Event const& event) {
                     << r.reason.value_or("") << "'.";
                 Notify(FDv2SourceResult{
                     FDv2SourceResult::Goodbye{r.reason, false}});
+                // Drop the current connection and reconnect; the protocol
+                // handler is reset so the new connection starts in a clean
+                // state.
+                protocol_handler_.Reset();
+                std::lock_guard lock(mutex_);
+                if (sse_client_) {
+                    sse_client_->async_restart("FDv2 goodbye received");
+                }
             } else if constexpr (std::is_same_v<T,
                                                 FDv2ProtocolHandler::Error>) {
                 if (r.kind == FDv2ProtocolHandler::Error::Kind::kServerError) {
diff --git a/libs/server-sdk/tests/change_notifier_test.cpp b/libs/server-sdk/tests/change_notifier_test.cpp
diff --git a/libs/server-sdk/tests/fdv2_data_system_test.cpp b/libs/server-sdk/tests/fdv2_data_system_test.cpp
