fdv2_data_system.hpp

#pragma once
 
#include "../../data_components/change_notifier/change_notifier.hpp"
#include "../../data_components/memory_store/memory_store.hpp"
#include "../../data_components/status_notifications/data_source_status_manager.hpp"
#include "../../data_interfaces/source/ifdv2_condition.hpp"
#include "../../data_interfaces/source/ifdv2_initializer_factory.hpp"
#include "../../data_interfaces/source/ifdv2_synchronizer_factory.hpp"
#include "../../data_interfaces/system/idata_system.hpp"
#include "conditions.hpp"
#include "source_manager.hpp"
 
#include <launchdarkly/async/cancellation.hpp>
#include <launchdarkly/data_model/selector.hpp>
#include <launchdarkly/logging/logger.hpp>
 
#include <boost/asio/any_io_executor.hpp>
 
#include <atomic>
#include <chrono>
#include <cstddef>
#include <memory>
#include <mutex>
#include <vector>
 
namespace launchdarkly::server_side::data_systems {
 
class FDv2DataSystem final : public data_interfaces::IDataSystem {
   public:
    FDv2DataSystem(
        std::vector<std::unique_ptr<data_interfaces::IFDv2InitializerFactory>>
            initializer_factories,
        std::vector<std::unique_ptr<data_interfaces::IFDv2SynchronizerFactory>>
            synchronizer_factories,
        std::unique_ptr<data_interfaces::IFDv2ConditionFactory>
            fallback_condition_factory,
        std::unique_ptr<data_interfaces::IFDv2ConditionFactory>
            recovery_condition_factory,
        boost::asio::any_io_executor ioc,
        data_components::DataSourceStatusManager* status_manager,
        Logger const& logger);
 
    ~FDv2DataSystem() override;
 
    FDv2DataSystem(FDv2DataSystem const&) = delete;
    FDv2DataSystem(FDv2DataSystem&&) = delete;
    FDv2DataSystem& operator=(FDv2DataSystem const&) = delete;
    FDv2DataSystem& operator=(FDv2DataSystem&&) = delete;
 
    std::shared_ptr<data_model::FlagDescriptor> GetFlag(
        std::string const& key) const override;
 
    std::shared_ptr<data_model::SegmentDescriptor> GetSegment(
        std::string const& key) const override;
 
    std::unordered_map<std::string, std::shared_ptr<data_model::FlagDescriptor>>
    AllFlags() const override;
 
    std::unordered_map<std::string,
                       std::shared_ptr<data_model::SegmentDescriptor>>
    AllSegments() const override;
 
    std::string const& Identity() const override;
 
    void Initialize() override;
 
    bool Initialized() const override;
 
   private:
    void Close();
 
    // Orchestration-loop methods. Each step chains the next via Future::Then,
    // 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);
    void StartSynchronizers();
    void RunSynchronizerNext();
    void OnSynchronizerResult(data_interfaces::FDv2SourceResult result);
    void OnConditionFired(data_interfaces::IFDv2Condition::Type type);
 
    // Schedules an FDv2 recovery attempt after the given TTL. Called with
    // mutex_ held. TTL of 0 disables the recovery and is a no-op.
    void ScheduleFDv2RetryLocked(std::chrono::seconds ttl);
 
    // Invoked when the FDv1 fallback TTL expires. Switches the source list
    // back to FDv2 and restarts the synchronizer phase.
    void OnFDv1RetryTimer();
 
    // 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.
    void ApplyChangeSet(
        data_model::ChangeSet<data_interfaces::ChangeSetData> change_set);
 
    // Logger is itself thread-safe and cheap to copy.
    Logger logger_;
 
    // Immutable after construction.
    boost::asio::any_io_executor const ioc_;
    std::vector<std::unique_ptr<data_interfaces::IFDv2InitializerFactory>> const
        initializer_factories_;
    std::unique_ptr<data_interfaces::IFDv2ConditionFactory> const
        fallback_condition_factory_;
    std::unique_ptr<data_interfaces::IFDv2ConditionFactory> const
        recovery_condition_factory_;
    // Non-owning. Lifetime guaranteed by the caller (see constructor doc).
    data_components::DataSourceStatusManager* const status_manager_;
 
    // Internally synchronized.
    data_components::MemoryStore store_;
    // Holds references to store_; declared after it so destruction order is
    // safe.
    data_components::ChangeNotifier change_notifier_;
 
    // Set by Initialize() to detect repeat or concurrent calls.
    std::atomic_bool initialize_called_;
 
    // Suppresses consecutive "interrupted" logs from the active synchronizer.
    std::atomic_bool last_logged_synchronizer_interrupted_;
 
    // Orchestration state, guarded by mutex_.
    std::mutex mutex_;
    bool closed_;
    data_model::Selector selector_;
    std::size_t initializer_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_;
 
    // Cancelled in Close() to abort any pending FDv1 fallback retry delay.
    async::CancellationSource fdv1_fallback_retry_cancel_;
};
 
}  // namespace launchdarkly::server_side::data_systems