feat: translate FDv1 status changes to FDv2 results in FDv1AdapterSynchronizer

Author: beekld

libs/server-sdk/src/data_systems/fdv2/fdv1_adapter_synchronizer.cpp

@@ -5,6 +5,7 @@
 namespace launchdarkly::server_side::data_systems {
 
 using data_interfaces::FDv2SourceResult;
+using DataSourceState = DataSourceStatus::DataSourceState;
 
 // ----- State -----
 
@@ -132,9 +133,32 @@ std::string const& FDv1AdapterSynchronizer::ConvertingDestination::Identity()
 // ----- FDv1AdapterSynchronizer -----
 
 FDv1AdapterSynchronizer::FDv1AdapterSynchronizer(
-    std::unique_ptr<data_interfaces::IDataSynchronizer> fdv1_source)
+    std::unique_ptr<data_interfaces::IDataSynchronizer> fdv1_source,
+    data_components::DataSourceStatusManager* status_manager)
     : state_(std::make_shared<State>()),
       destination_(std::make_unique<ConvertingDestination>(state_)),
+      status_manager_(status_manager),
+      status_subscription_(status_manager_->OnDataSourceStatusChange(
+          [state = state_](DataSourceStatus status) {
+              auto error = status.LastError();
+              if (!error) {
+                  return;
+              }
+              switch (status.State()) {
+                  case DataSourceState::kInterrupted:
+                      state->Notify(FDv2SourceResult{
+                          FDv2SourceResult::Interrupted{*error}});
+                      break;
+                  case DataSourceState::kOff:
+                      state->Notify(FDv2SourceResult{
+                          FDv2SourceResult::TerminalError{*error}});
+                      break;
+                  case DataSourceState::kInitializing:
+                  case DataSourceState::kValid:
+                      // No FDv2 result for these states.
+                      break;
+              }
+          })),
       fdv1_source_(std::move(fdv1_source)) {}
 
 FDv1AdapterSynchronizer::~FDv1AdapterSynchronizer() {

libs/server-sdk/src/data_systems/fdv2/fdv1_adapter_synchronizer.hpp

@@ -1,10 +1,12 @@
 #pragma once
 
+#include "../../data_components/status_notifications/data_source_status_manager.hpp"
 #include "../../data_interfaces/destination/idestination.hpp"
 #include "../../data_interfaces/source/idata_synchronizer.hpp"
 #include "../../data_interfaces/source/ifdv2_synchronizer.hpp"
 
 #include <launchdarkly/async/promise.hpp>
+#include <launchdarkly/connection.hpp>
 
 #include <deque>
 #include <memory>
@@ -23,15 +25,26 @@ namespace launchdarkly::server_side::data_systems {
  * and fdv1_fallback = false (the directive does not re-fire from FDv1 data).
  *
  * Threading: Next() and Close() may be called from any thread; only one
- * Next() may be outstanding at a time. The adapter blocks in its destructor
- * waiting for the FDv1 source's ShutdownAsync completion, so no callbacks
- * are in flight when the wrapped source is destroyed.
+ * Next() may be outstanding at a time. Member declaration order ensures
+ * the wrapped FDv1 source destructs before destination_ and state_, so any
+ * in-flight FDv1 callbacks land on live objects during teardown. This
+ * relies on the wrapped IDataSynchronizer blocking on its in-flight work
+ * in its destructor.
  */
 class FDv1AdapterSynchronizer final
     : public data_interfaces::IFDv2Synchronizer {
    public:
-    explicit FDv1AdapterSynchronizer(
-        std::unique_ptr<data_interfaces::IDataSynchronizer> fdv1_source);
+    /**
+     * @param fdv1_source The wrapped source. Must have been constructed
+     *                    with status_manager as its status sink so that
+     *                    state changes flow back into this adapter.
+     * @param status_manager Non-owning. The caller retains ownership and
+     *                       must keep it alive for the lifetime of the
+     *                       wrapped source and this adapter.
+     */
+    FDv1AdapterSynchronizer(
+        std::unique_ptr<data_interfaces::IDataSynchronizer> fdv1_source,
+        data_components::DataSourceStatusManager* status_manager);
 
     ~FDv1AdapterSynchronizer() override;
 
@@ -95,9 +108,17 @@ class FDv1AdapterSynchronizer final
         std::weak_ptr<State> state_;
     };
 
-    // const after construction.
+    // shared_ptr so async callbacks that may fire after this is destroyed
+    // can hold their own reference.
     std::shared_ptr<State> const state_;
     std::unique_ptr<ConvertingDestination> const destination_;
+
+    // Non-owning. The caller must keep this alive for the lifetime of
+    // the wrapped source, which holds a reference to it for status
+    // reporting.
+    data_components::DataSourceStatusManager* const status_manager_;
+    std::unique_ptr<IConnection> const status_subscription_;
+
     std::unique_ptr<data_interfaces::IDataSynchronizer> const fdv1_source_;
 
     // Thread-safe primitive.

libs/server-sdk/tests/fdv1_adapter_synchronizer_test.cpp

@@ -8,6 +8,8 @@
 #include <utility>
 
 using namespace launchdarkly;
+using namespace launchdarkly::server_side;
+using namespace launchdarkly::server_side::data_components;
 using namespace launchdarkly::server_side::data_interfaces;
 using namespace launchdarkly::server_side::data_systems;
 using namespace std::chrono_literals;
@@ -18,6 +20,8 @@ namespace {
 // the IDestination it was given so the test can drive Init/Upsert.
 class MockFDv1Source final : public IDataSynchronizer {
    public:
+    explicit MockFDv1Source(DataSourceStatusManager& /*status_manager*/) {}
+
     void StartAsync(IDestination* destination,
                     data_model::SDKDataSet const* bootstrap) override {
         ++start_count;
@@ -46,9 +50,10 @@ class MockFDv1Source final : public IDataSynchronizer {
 }  // namespace
 
 TEST(FDv1AdapterSynchronizerTest, FirstNextStartsFDv1Source) {
-    auto source = std::make_unique<MockFDv1Source>();
+    DataSourceStatusManager status_manager;
+    auto source = std::make_unique<MockFDv1Source>(status_manager);
     auto* source_ptr = source.get();
-    FDv1AdapterSynchronizer adapter(std::move(source));
+    FDv1AdapterSynchronizer adapter(std::move(source), &status_manager);
 
     auto future = adapter.Next(data_model::Selector{});
 
@@ -58,9 +63,10 @@ TEST(FDv1AdapterSynchronizerTest, FirstNextStartsFDv1Source) {
 }
 
 TEST(FDv1AdapterSynchronizerTest, SecondNextDoesNotRestartSource) {
-    auto source = std::make_unique<MockFDv1Source>();
+    DataSourceStatusManager status_manager;
+    auto source = std::make_unique<MockFDv1Source>(status_manager);
     auto* source_ptr = source.get();
-    FDv1AdapterSynchronizer adapter(std::move(source));
+    FDv1AdapterSynchronizer adapter(std::move(source), &status_manager);
 
     auto first = adapter.Next(data_model::Selector{});
     source_ptr->destination_->Init(data_model::SDKDataSet{});
@@ -72,9 +78,10 @@ TEST(FDv1AdapterSynchronizerTest, SecondNextDoesNotRestartSource) {
 }
 
 TEST(FDv1AdapterSynchronizerTest, FDv1InitProducesFullChangeSet) {
-    auto source = std::make_unique<MockFDv1Source>();
+    DataSourceStatusManager status_manager;
+    auto source = std::make_unique<MockFDv1Source>(status_manager);
     auto* source_ptr = source.get();
-    FDv1AdapterSynchronizer adapter(std::move(source));
+    FDv1AdapterSynchronizer adapter(std::move(source), &status_manager);
 
     auto future = adapter.Next(data_model::Selector{});
 
@@ -97,9 +104,10 @@ TEST(FDv1AdapterSynchronizerTest, FDv1InitProducesFullChangeSet) {
 }
 
 TEST(FDv1AdapterSynchronizerTest, FDv1UpsertProducesPartialChangeSet) {
-    auto source = std::make_unique<MockFDv1Source>();
+    DataSourceStatusManager status_manager;
+    auto source = std::make_unique<MockFDv1Source>(status_manager);
     auto* source_ptr = source.get();
-    FDv1AdapterSynchronizer adapter(std::move(source));
+    FDv1AdapterSynchronizer adapter(std::move(source), &status_manager);
 
     // Flag upsert.
     auto flag_future = adapter.Next(data_model::Selector{});
@@ -138,8 +146,9 @@ TEST(FDv1AdapterSynchronizerTest, FDv1UpsertProducesPartialChangeSet) {
 }
 
 TEST(FDv1AdapterSynchronizerTest, ClosePendingNextReturnsShutdown) {
-    auto source = std::make_unique<MockFDv1Source>();
-    FDv1AdapterSynchronizer adapter(std::move(source));
+    DataSourceStatusManager status_manager;
+    auto source = std::make_unique<MockFDv1Source>(status_manager);
+    FDv1AdapterSynchronizer adapter(std::move(source), &status_manager);
 
     auto future = adapter.Next(data_model::Selector{});
     EXPECT_FALSE(future.IsFinished());
@@ -153,9 +162,10 @@ TEST(FDv1AdapterSynchronizerTest, ClosePendingNextReturnsShutdown) {
 }
 
 TEST(FDv1AdapterSynchronizerTest, CloseShutsDownStartedFDv1Source) {
-    auto source = std::make_unique<MockFDv1Source>();
+    DataSourceStatusManager status_manager;
+    auto source = std::make_unique<MockFDv1Source>(status_manager);
     auto* source_ptr = source.get();
-    FDv1AdapterSynchronizer adapter(std::move(source));
+    FDv1AdapterSynchronizer adapter(std::move(source), &status_manager);
 
     adapter.Next(data_model::Selector{});
     adapter.Close();
@@ -164,9 +174,10 @@ TEST(FDv1AdapterSynchronizerTest, CloseShutsDownStartedFDv1Source) {
 }
 
 TEST(FDv1AdapterSynchronizerTest, CloseWithoutStartDoesNotShutDownFDv1Source) {
-    auto source = std::make_unique<MockFDv1Source>();
+    DataSourceStatusManager status_manager;
+    auto source = std::make_unique<MockFDv1Source>(status_manager);
     auto* source_ptr = source.get();
-    FDv1AdapterSynchronizer adapter(std::move(source));
+    FDv1AdapterSynchronizer adapter(std::move(source), &status_manager);
 
     // No Next() call — FDv1 source was never started.
     adapter.Close();
@@ -175,9 +186,85 @@ TEST(FDv1AdapterSynchronizerTest, CloseWithoutStartDoesNotShutDownFDv1Source) {
     EXPECT_EQ(0, source_ptr->shutdown_count);
 }
 
+TEST(FDv1AdapterSynchronizerTest, QueuedResultsDrainInFifoOrder) {
+    DataSourceStatusManager status_manager;
+    auto source = std::make_unique<MockFDv1Source>(status_manager);
+    auto* source_ptr = source.get();
+    FDv1AdapterSynchronizer adapter(std::move(source), &status_manager);
+
+    // Start the source by satisfying one Next() with an Init.
+    auto first = adapter.Next(data_model::Selector{});
+    source_ptr->destination_->Init(data_model::SDKDataSet{});
+    first.WaitForResult(1s);
+
+    // Two upserts queue with no Next() in flight.
+    data_model::Flag flag_a;
+    flag_a.key = "a";
+    data_model::Flag flag_b;
+    flag_b.key = "b";
+    source_ptr->destination_->Upsert("a", data_model::FlagDescriptor(flag_a));
+    source_ptr->destination_->Upsert("b", data_model::FlagDescriptor(flag_b));
+
+    // Drain in FIFO order.
+    auto r1 = adapter.Next(data_model::Selector{}).WaitForResult(1s);
+    auto r2 = adapter.Next(data_model::Selector{}).WaitForResult(1s);
+    ASSERT_TRUE(r1.has_value());
+    ASSERT_TRUE(r2.has_value());
+    auto* cs1 = std::get_if<FDv2SourceResult::ChangeSet>(&r1->value);
+    auto* cs2 = std::get_if<FDv2SourceResult::ChangeSet>(&r2->value);
+    ASSERT_NE(cs1, nullptr);
+    ASSERT_NE(cs2, nullptr);
+    ASSERT_EQ(1u, cs1->change_set.data.size());
+    ASSERT_EQ(1u, cs2->change_set.data.size());
+    EXPECT_EQ("a", cs1->change_set.data[0].key);
+    EXPECT_EQ("b", cs2->change_set.data[0].key);
+}
+
+TEST(FDv1AdapterSynchronizerTest, InterruptedStatusProducesInterruptedResult) {
+    DataSourceStatusManager status_manager;
+    auto source = std::make_unique<MockFDv1Source>(status_manager);
+    FDv1AdapterSynchronizer adapter(std::move(source), &status_manager);
+
+    // kInterrupted from kInitializing stays kInitializing; drive past first.
+    status_manager.SetState(DataSourceStatus::DataSourceState::kValid);
+
+    auto future = adapter.Next(data_model::Selector{});
+    status_manager.SetState(
+        DataSourceStatus::DataSourceState::kInterrupted,
+        DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError, "boom");
+
+    auto result = future.WaitForResult(1s);
+    ASSERT_TRUE(result.has_value());
+    auto* interrupted =
+        std::get_if<FDv2SourceResult::Interrupted>(&result->value);
+    ASSERT_NE(interrupted, nullptr);
+    EXPECT_EQ(DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError,
+              interrupted->error.Kind());
+}
+
+TEST(FDv1AdapterSynchronizerTest, OffStatusProducesTerminalErrorResult) {
+    DataSourceStatusManager status_manager;
+    auto source = std::make_unique<MockFDv1Source>(status_manager);
+    FDv1AdapterSynchronizer adapter(std::move(source), &status_manager);
+
+    auto future = adapter.Next(data_model::Selector{});
+    status_manager.SetState(
+        DataSourceStatus::DataSourceState::kOff,
+        DataSourceStatus::ErrorInfo::ErrorKind::kErrorResponse, "401");
+
+    auto result = future.WaitForResult(1s);
+    ASSERT_TRUE(result.has_value());
+    auto* terminal =
+        std::get_if<FDv2SourceResult::TerminalError>(&result->value);
+    ASSERT_NE(terminal, nullptr);
+    EXPECT_EQ(DataSourceStatus::ErrorInfo::ErrorKind::kErrorResponse,
+              terminal->error.Kind());
+}
+
 TEST(FDv1AdapterSynchronizerTest, NextAfterCloseReturnsShutdown) {
-    auto source = std::make_unique<MockFDv1Source>();
-    FDv1AdapterSynchronizer adapter(std::move(source));
+    DataSourceStatusManager status_manager;
+    auto source = std::make_unique<MockFDv1Source>(status_manager);
+    FDv1AdapterSynchronizer adapter(std::move(source), &status_manager);
 
     adapter.Close();
     auto future = adapter.Next(data_model::Selector{});