feat: Honor the service's FDv1 fallback directive in the client

libs/client-sdk/src/CMakeLists.txt

@@ -23,6 +23,7 @@ target_sources(${LIBNAME} PRIVATE
         data_sources/fdv2/streaming_synchronizer.cpp
         data_sources/fdv2/fdv2_data_source.cpp
         data_sources/fdv2/cache_initializer.cpp
+        data_sources/fdv2/fdv1_adapter_synchronizer.cpp
         data_sources/data_source_event_handler.cpp
         data_sources/polling_data_source.cpp
         flag_manager/flag_store.cpp
@@ -51,6 +52,7 @@ target_sources(${LIBNAME} PRIVATE
         data_sources/fdv2/streaming_synchronizer.hpp
         data_sources/fdv2/fdv2_data_source.hpp
         data_sources/fdv2/cache_initializer.hpp
+        data_sources/fdv2/fdv1_adapter_synchronizer.hpp
         flag_manager/flag_store.hpp
         flag_manager/flag_updater.hpp
         bindings/c/sdk.cpp

libs/client-sdk/src/data_sources/fdv2/fdv1_adapter_synchronizer.cpp

@@ -0,0 +1,204 @@
+#include "fdv1_adapter_synchronizer.hpp"
+
+#include <utility>
+
+namespace launchdarkly::client_side::data_sources {
+
+using DataSourceState = DataSourceStatus::DataSourceState;
+
+// ----- State -----
+
+FDv1AdapterSynchronizer::State::State(async::Future<std::monostate> closed)
+    : closed_(std::move(closed)) {}
+
+async::Future<FDv2SourceResult> FDv1AdapterSynchronizer::State::GetNext() {
+    std::lock_guard lock(mutex_);
+    if (!result_queue_.empty()) {
+        auto result = std::move(result_queue_.front());
+        result_queue_.pop_front();
+        return async::MakeFuture(std::move(result));
+    }
+    return pending_promise_.emplace().GetFuture();
+}
+
+void FDv1AdapterSynchronizer::State::ResolvePendingAsShutdown() {
+    std::optional<async::Promise<FDv2SourceResult>> promise;
+    {
+        std::lock_guard lock(mutex_);
+        if (pending_promise_) {
+            promise = std::move(pending_promise_);
+            pending_promise_.reset();
+        }
+    }
+    if (promise) {
+        promise->Resolve(FDv2SourceResult{FDv2SourceResult::Shutdown{}});
+    }
+}
+
+void FDv1AdapterSynchronizer::State::Notify(FDv2SourceResult result) {
+    std::optional<async::Promise<FDv2SourceResult>> promise;
+    {
+        std::lock_guard lock(mutex_);
+        if (closed_.IsFinished()) {
+            return;
+        }
+        if (pending_promise_) {
+            promise = std::move(pending_promise_);
+            pending_promise_.reset();
+        } else {
+            result_queue_.push_back(std::move(result));
+            return;
+        }
+    }
+    // Resolve outside the lock. Promise::Resolve may invoke inline
+    // continuations that could call back into Notify or GetNext.
+    promise->Resolve(std::move(result));
+}
+
+// ----- ConvertingSink -----
+
+FDv1AdapterSynchronizer::ConvertingSink::ConvertingSink(
+    std::weak_ptr<State> state)
+    : state_(std::move(state)) {}
+
+void FDv1AdapterSynchronizer::ConvertingSink::Init(
+    Context const& /* context */,
+    std::unordered_map<std::string, ItemDescriptor> data) {
+    auto state = state_.lock();
+    if (!state) {
+        return;
+    }
+    FlagChangeSetData changes;
+    changes.reserve(data.size());
+    for (auto& [key, item] : data) {
+        changes.push_back(FlagChange{key, std::move(item)});
+    }
+    state->Notify(FDv2SourceResult{FDv2SourceResult::ChangeSet{
+        FlagChangeSet{data_model::ChangeSetType::kFull, std::move(changes),
+                      data_model::Selector{}}}});
+}
+
+void FDv1AdapterSynchronizer::ConvertingSink::Upsert(
+    Context const& /* context */,
+    std::string key,
+    ItemDescriptor item) {
+    auto state = state_.lock();
+    if (!state) {
+        return;
+    }
+    state->Notify(FDv2SourceResult{FDv2SourceResult::ChangeSet{
+        FlagChangeSet{data_model::ChangeSetType::kPartial,
+                      {FlagChange{std::move(key), std::move(item)}},
+                      data_model::Selector{}}}});
+}
+
+void FDv1AdapterSynchronizer::ConvertingSink::Apply(
+    Context const& /* context */,
+    FlagChangeSet change_set,
+    bool /* from_cache */) {
+    auto state = state_.lock();
+    if (!state) {
+        return;
+    }
+    state->Notify(
+        FDv2SourceResult{FDv2SourceResult::ChangeSet{std::move(change_set)}});
+}
+
+// ----- FDv1AdapterSynchronizer -----
+
+namespace {
+
+// Turns the wrapped source's status into a result, or into nothing when the
+// status carries no error.
+std::optional<FDv2SourceResult> ResultForStatus(
+    DataSourceStatus const& status) {
+    auto const error = status.LastError();
+    if (!error) {
+        return std::nullopt;
+    }
+    switch (status.State()) {
+        case DataSourceState::kInterrupted:
+        // An error before the source ever became valid is reported as still
+        // initializing, but it is the same recoverable failure.
+        case DataSourceState::kInitializing:
+            return FDv2SourceResult{FDv2SourceResult::Interrupted{*error}};
+        case DataSourceState::kShutdown:
+            return FDv2SourceResult{FDv2SourceResult::TerminalError{*error}};
+        case DataSourceState::kValid:
+        case DataSourceState::kSetOffline:
+            return std::nullopt;
+    }
+    return std::nullopt;
+}
+
+}  // namespace
+
+FDv1AdapterSynchronizer::FDv1AdapterSynchronizer(SourceBuilder source_builder)
+    : state_(std::make_shared<State>(close_promise_.GetFuture())),
+      sink_(std::make_shared<ConvertingSink>(state_)),
+      status_manager_(std::make_shared<DataSourceStatusManager>()),
+      status_subscription_(status_manager_->OnDataSourceStatusChange(
+          [state = state_](DataSourceStatus status) {
+              if (auto result = ResultForStatus(status)) {
+                  state->Notify(std::move(*result));
+              }
+          })),
+      fdv1_source_(source_builder(sink_.get(), status_manager_.get())) {}
+
+FDv1AdapterSynchronizer::~FDv1AdapterSynchronizer() {
+    Close();
+}
+
+async::Future<FDv2SourceResult> FDv1AdapterSynchronizer::Next(
+    data_model::Selector /* selector */) {
+    auto closed = close_promise_.GetFuture();
+    if (closed.IsFinished()) {
+        return async::MakeFuture(
+            FDv2SourceResult{FDv2SourceResult::Shutdown{}});
+    }
+    {
+        std::lock_guard lock(lifecycle_mutex_);
+        if (!started_) {
+            started_ = true;
+            fdv1_source_->Start();
+        }
+    }
+    auto result_future = state_->GetNext();
+    if (result_future.IsFinished()) {
+        return result_future;
+    }
+    return async::WhenAny(closed, result_future)
+        .Then(
+            [state = state_, result_future](std::size_t const& idx) mutable
+                -> async::Future<FDv2SourceResult> {
+                if (idx == 0) {
+                    state->ResolvePendingAsShutdown();
+                    return async::MakeFuture(
+                        FDv2SourceResult{FDv2SourceResult::Shutdown{}});
+                }
+                return result_future;
+            },
+            async::kInlineExecutor);
+}
+
+void FDv1AdapterSynchronizer::Close() {
+    if (!close_promise_.Resolve(std::monostate{})) {
+        return;
+    }
+    std::lock_guard lock(lifecycle_mutex_);
+    bool const was_started = started_;
+    started_ = true;
+    if (was_started) {
+        // The completion does no work. It exists to hold these alive until
+        // shutdown finishes.
+        fdv1_source_->ShutdownAsync(
+            [sink = sink_, status = status_manager_, source = fdv1_source_] {});
+    }
+}
+
+std::string const& FDv1AdapterSynchronizer::Identity() const {
+    static std::string const identity = "FDv1 fallback adapter";
+    return identity;
+}
+
+}  // namespace launchdarkly::client_side::data_sources

libs/client-sdk/src/data_sources/fdv2/fdv1_adapter_synchronizer.hpp

@@ -0,0 +1,123 @@
+#pragma once
+
+#include "../data_source.hpp"
+#include "../data_source_status_manager.hpp"
+#include "../data_source_update_sink.hpp"
+#include "ifdv2_synchronizer.hpp"
+
+#include <launchdarkly/async/promise.hpp>
+#include <launchdarkly/connection.hpp>
+#include <launchdarkly/context.hpp>
+
+#include <deque>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string>
+
+namespace launchdarkly::client_side::data_sources {
+
+/**
+ * Presents an FDv1 data source as an FDv2 synchronizer.
+ *
+ * FDv1 has no selectors, so the changesets this reports carry none.
+ *
+ * Thread safety: Next() and Close() may be called from any thread. Only one
+ * Next() may be outstanding at a time.
+ */
+class FDv1AdapterSynchronizer final : public IFDv2Synchronizer {
+   public:
+    /**
+     * Builds the wrapped FDv1 source. Called once during construction, with a
+     * sink and status manager that stay valid for the source's lifetime.
+     */
+    using SourceBuilder =
+        std::function<std::shared_ptr<IDataSource>(IDataSourceUpdateSink*,
+                                                   DataSourceStatusManager*)>;
+
+    explicit FDv1AdapterSynchronizer(SourceBuilder source_builder);
+
+    ~FDv1AdapterSynchronizer() override;
+
+    async::Future<FDv2SourceResult> Next(
+        data_model::Selector selector) override;
+
+    void Close() override;
+
+    [[nodiscard]] std::string const& Identity() const override;
+
+   private:
+    /**
+     * Holds the result queue and the pending Next() promise. Shared with the
+     * wrapped source's sink and status subscription. Thread-safe.
+     */
+    class State {
+       public:
+        explicit State(async::Future<std::monostate> closed);
+
+        async::Future<FDv2SourceResult> GetNext();
+
+        /** Resolves any pending Next() with Shutdown and clears it. */
+        void ResolvePendingAsShutdown();
+
+        /**
+         * Hands the result to a pending Next(), or queues it for the next
+         * one. Results that arrive after Close() are dropped.
+         */
+        void Notify(FDv2SourceResult result);
+
+       private:
+        // Finished once the owning adapter's Close() has run.
+        async::Future<std::monostate> const closed_;
+
+        mutable std::mutex mutex_;
+        // Both protected by mutex_.
+        std::optional<async::Promise<FDv2SourceResult>> pending_promise_;
+        std::deque<FDv2SourceResult> result_queue_;
+    };
+
+    /**
+     * Turns the FDv1 source's Init and Upsert calls into FDv2 changesets
+     * queued on State. Thread-safe (delegates to State).
+     */
+    class ConvertingSink final : public IDataSourceUpdateSink {
+       public:
+        explicit ConvertingSink(std::weak_ptr<State> state);
+
+        void Init(
+            Context const& context,
+            std::unordered_map<std::string, ItemDescriptor> data) override;
+        void Upsert(Context const& context,
+                    std::string key,
+                    ItemDescriptor item) override;
+        void Apply(Context const& context,
+                   FlagChangeSet change_set,
+                   bool from_cache) override;
+
+       private:
+        std::weak_ptr<State> state_;
+    };
+
+    // Thread-safe primitive. Declared before state_ so state_'s constructor
+    // can take a future from it.
+    async::Promise<std::monostate> close_promise_;
+
+    // shared_ptr so async callbacks that fire after this is destroyed can
+    // hold their own reference.
+    std::shared_ptr<State> const state_;
+    std::shared_ptr<ConvertingSink> const sink_;
+    std::shared_ptr<DataSourceStatusManager> const status_manager_;
+    std::unique_ptr<IConnection> const status_subscription_;
+
+    std::shared_ptr<IDataSource> const fdv1_source_;
+
+    // Serializes Start and ShutdownAsync on fdv1_source_ across concurrent
+    // Next() and Close() calls.
+    std::mutex lifecycle_mutex_;
+    // Protected by lifecycle_mutex_. Set when Next() starts the source, or
+    // when Close() runs first, so that a later Next() cannot start it.
+    bool started_ = false;
+};
+
+}  // namespace launchdarkly::client_side::data_sources