feat: add FDv1AdapterSynchronizer wrapping IDataSynchronizer as IFDv2Synchronizer (#540)

## Summary

Adds an adapter that wraps any FDv1 `IDataSynchronizer` (e.g. the
existing `StreamingDataSource`) and presents it as an
`IFDv2Synchronizer`. FDv1 `Init`/`Upsert` callbacks delivered through an
internal `IDestination` are translated into
`FDv2SourceResult::ChangeSet` results (`kFull` for `Init`, `kPartial`
for `Upsert`), with empty selectors and `fdv1_fallback=false`.

The orchestrator picks this up after `SwitchToFDv1Fallback()` once a
factory whose `IsFDv1Fallback()` returns true is configured. Wiring the
adapter into the public config builders is a separate concern, deferred
until the Configuration API work.

## Test plan

- [x] 8 unit tests cover: lazy start, no-restart on second Next,
Init→kFull, Upsert→kPartial (both overloads),
pending-Next-resolved-as-Shutdown-on-close, ShutdownAsync called once on
close (and not at all if never started), Next-after-Close
- [x] Full server suite green (486 tests)


<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Touches core flag data sync and fallback behavior with async lifecycle
and threading; risk is mitigated by isolation behind IFDv2Synchronizer
and broad unit tests, but mis-mapping status or shutdown could affect
live updates until config is wired.
> 
> **Overview**
> Adds **`FDv1AdapterSynchronizer`**, which wraps an FDv1
`IDataSynchronizer` (e.g. streaming) and exposes it as
**`IFDv2Synchronizer`** for FDv2 orchestration and FDv1 fallback paths.
> 
> An internal **`ConvertingDestination`** turns FDv1 `Init`/`Upsert`
into **`FDv2SourceResult::ChangeSet`** (`kFull` / `kPartial`, empty
selectors, **`fdv1_fallback=false`**). **`Next()`** lazily starts the
wrapped source once, queues multiple results in FIFO order, and races
close vs. pending work via **`WhenAny`**. **`DataSourceStatusManager`**
errors map to **`Interrupted`** or **`TerminalError`**.
> 
> The new sources are registered in **`CMakeLists.txt`**. **12 gtest
cases** cover lifecycle, conversion, queuing, status mapping, and
shutdown. Public config wiring is left for a follow-up.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
796d25c. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: beekld

libs/server-sdk/src/CMakeLists.txt

@@ -76,6 +76,8 @@ target_sources(${LIBNAME}
         data_systems/fdv2/source_manager.cpp
         data_systems/fdv2/fdv2_data_system.hpp
         data_systems/fdv2/fdv2_data_system.cpp
+        data_systems/fdv2/fdv1_adapter_synchronizer.hpp
+        data_systems/fdv2/fdv1_adapter_synchronizer.cpp
         data_systems/background_sync/sources/streaming/streaming_data_source.hpp
         data_systems/background_sync/sources/streaming/streaming_data_source.cpp
         data_systems/background_sync/sources/streaming/event_handler.hpp

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

@@ -0,0 +1,208 @@
+#include "fdv1_adapter_synchronizer.hpp"
+
+#include <utility>
+
+namespace launchdarkly::server_side::data_systems {
+
+using data_interfaces::FDv2SourceResult;
+using DataSourceState = DataSourceStatus::DataSourceState;
+
+// ----- State -----
+
+FDv1AdapterSynchronizer::State::State(
+    async::Future<std::monostate> closed_future)
+    : closed_future_(std::move(closed_future)) {}
+
+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_future_.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));
+}
+
+// ----- ConvertingDestination -----
+
+FDv1AdapterSynchronizer::ConvertingDestination::ConvertingDestination(
+    std::weak_ptr<State> state)
+    : state_(std::move(state)) {}
+
+void FDv1AdapterSynchronizer::ConvertingDestination::Init(
+    data_model::SDKDataSet data_set) {
+    auto state = state_.lock();
+    if (!state) {
+        return;
+    }
+    data_interfaces::ChangeSetData changes;
+    changes.reserve(data_set.flags.size() + data_set.segments.size());
+    for (auto& [key, flag] : data_set.flags) {
+        changes.push_back({key, std::move(flag)});
+    }
+    for (auto& [key, segment] : data_set.segments) {
+        changes.push_back({key, std::move(segment)});
+    }
+    state->Notify(FDv2SourceResult{FDv2SourceResult::ChangeSet{
+        data_model::ChangeSet<data_interfaces::ChangeSetData>{
+            data_model::ChangeSetType::kFull, std::move(changes),
+            data_model::Selector{}}}});
+}
+
+void FDv1AdapterSynchronizer::ConvertingDestination::Upsert(
+    std::string const& key,
+    data_model::FlagDescriptor flag) {
+    auto state = state_.lock();
+    if (!state) {
+        return;
+    }
+    data_interfaces::ChangeSetData changes;
+    changes.push_back({key, std::move(flag)});
+    state->Notify(FDv2SourceResult{FDv2SourceResult::ChangeSet{
+        data_model::ChangeSet<data_interfaces::ChangeSetData>{
+            data_model::ChangeSetType::kPartial, std::move(changes),
+            data_model::Selector{}}}});
+}
+
+void FDv1AdapterSynchronizer::ConvertingDestination::Upsert(
+    std::string const& key,
+    data_model::SegmentDescriptor segment) {
+    auto state = state_.lock();
+    if (!state) {
+        return;
+    }
+    data_interfaces::ChangeSetData changes;
+    changes.push_back({key, std::move(segment)});
+    state->Notify(FDv2SourceResult{FDv2SourceResult::ChangeSet{
+        data_model::ChangeSet<data_interfaces::ChangeSetData>{
+            data_model::ChangeSetType::kPartial, std::move(changes),
+            data_model::Selector{}}}});
+}
+
+std::string const& FDv1AdapterSynchronizer::ConvertingDestination::Identity()
+    const {
+    static std::string const identity = "FDv1 adapter destination";
+    return identity;
+}
+
+// ----- FDv1AdapterSynchronizer -----
+
+FDv1AdapterSynchronizer::FDv1AdapterSynchronizer(SourceBuilder source_builder)
+    : state_(std::make_shared<State>(close_promise_.GetFuture())),
+      destination_(std::make_unique<ConvertingDestination>(state_)),
+      status_manager_(
+          std::make_unique<data_components::DataSourceStatusManager>()),
+      status_subscription_(status_manager_->OnDataSourceStatusChange(
+          [state = state_](DataSourceStatus status) {
+              auto error = status.LastError();
+              if (!error) {
+                  return;
+              }
+              switch (status.State()) {
+                  case DataSourceState::kInterrupted:
+                  case DataSourceState::kInitializing:
+                      // Recoverable error.
+                      state->Notify(FDv2SourceResult{
+                          FDv2SourceResult::Interrupted{*error}});
+                      break;
+                  case DataSourceState::kOff:
+                      // Terminal error.
+                      state->Notify(FDv2SourceResult{
+                          FDv2SourceResult::TerminalError{*error}});
+                      break;
+                  case DataSourceState::kValid:
+                      // Recovery; no FDv2 result.
+                      break;
+              }
+          })),
+      fdv1_source_(source_builder(*status_manager_)) {}
+
+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_->StartAsync(destination_.get(),
+                                     /*bootstrap_data=*/nullptr);
+        }
+    }
+    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) {
+        fdv1_source_->ShutdownAsync([] {});
+    }
+}
+
+std::string const& FDv1AdapterSynchronizer::Identity() const {
+    static std::string const identity = "FDv1 fallback adapter";
+    return identity;
+}
+
+}  // namespace launchdarkly::server_side::data_systems

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

@@ -0,0 +1,130 @@
+#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 <functional>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <variant>
+
+namespace launchdarkly::server_side::data_systems {
+
+/**
+ * Adapts an FDv1 IDataSynchronizer to the IFDv2Synchronizer interface.
+ *
+ * FDv1 Init/Upsert callbacks delivered through an internal IDestination are
+ * translated into FDv2SourceResult::ChangeSet results, with empty selectors
+ * 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. 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:
+    using SourceBuilder =
+        std::function<std::shared_ptr<data_interfaces::IDataSynchronizer>(
+            data_components::DataSourceStatusManager&)>;
+
+    /**
+     * @param source_builder Called once during construction with the
+     *                       adapter's status manager. Returns the wrapped
+     *                       FDv1 source, which must be constructed against
+     *                       the provided manager as its status sink.
+     */
+    explicit FDv1AdapterSynchronizer(SourceBuilder source_builder);
+
+    ~FDv1AdapterSynchronizer() override;
+
+    async::Future<data_interfaces::FDv2SourceResult> Next(
+        data_model::Selector selector) override;
+    void Close() override;
+    [[nodiscard]] std::string const& Identity() const override;
+
+   private:
+    /**
+     * Holds the result queue and pending Next() promise; shared with the
+     * FDv1 source's IDestination via the inner ConvertingDestination.
+     * All methods are thread-safe.
+     */
+    class State {
+       public:
+        explicit State(async::Future<std::monostate> closed_future);
+
+        async::Future<data_interfaces::FDv2SourceResult> GetNext();
+
+        // Resolves any pending Next() promise with Shutdown and clears it.
+        // Called on the close path so the abandoned promise doesn't leave
+        // potential continuations dangling.
+        void ResolvePendingAsShutdown();
+
+        void Notify(data_interfaces::FDv2SourceResult result);
+
+       private:
+        // Finished once the owning FDv1AdapterSynchronizer's close_promise_
+        // is resolved. Read in Notify to drop late results.
+        async::Future<std::monostate> const closed_future_;
+
+        mutable std::mutex mutex_;
+        // Protected by mutex_.
+        std::optional<async::Promise<data_interfaces::FDv2SourceResult>>
+            pending_promise_;
+        std::deque<data_interfaces::FDv2SourceResult> result_queue_;
+    };
+
+    /**
+     * Translates FDv1 IDestination callbacks into FDv2 results queued on
+     * State. Thread-safe (delegates to State).
+     */
+    class ConvertingDestination final : public data_interfaces::IDestination {
+       public:
+        explicit ConvertingDestination(std::weak_ptr<State> state);
+        void Init(data_model::SDKDataSet data_set) override;
+        void Upsert(std::string const& key,
+                    data_model::FlagDescriptor flag) override;
+        void Upsert(std::string const& key,
+                    data_model::SegmentDescriptor segment) override;
+        [[nodiscard]] std::string const& Identity() const 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 may fire after this is destroyed
+    // can hold their own reference.
+    std::shared_ptr<State> const state_;
+    std::unique_ptr<ConvertingDestination> const destination_;
+
+    std::unique_ptr<data_components::DataSourceStatusManager> const
+        status_manager_;
+    std::unique_ptr<IConnection> const status_subscription_;
+
+    std::shared_ptr<data_interfaces::IDataSynchronizer> const fdv1_source_;
+
+    // Serializes StartAsync and ShutdownAsync on fdv1_source_ across
+    // concurrent Next() and Close() calls.
+    std::mutex lifecycle_mutex_;
+    // Protected by lifecycle_mutex_. Set when Next() calls StartAsync, or
+    // when Close() runs without a prior start (to gate any later Next()
+    // from calling StartAsync after Close).
+    bool started_ = false;
+};
+
+}  // namespace launchdarkly::server_side::data_systems