feat: Add the FDv2 cache initializer and per-context freshness to the client

libs/client-sdk/src/CMakeLists.txt

@@ -22,6 +22,7 @@ target_sources(${LIBNAME} PRIVATE
         data_sources/fdv2/polling_synchronizer.cpp
         data_sources/fdv2/streaming_synchronizer.cpp
         data_sources/fdv2/fdv2_data_source.cpp
+        data_sources/fdv2/cache_initializer.cpp
         data_sources/data_source_event_handler.cpp
         data_sources/polling_data_source.cpp
         flag_manager/flag_store.cpp
@@ -49,6 +50,7 @@ target_sources(${LIBNAME} PRIVATE
         data_sources/fdv2/polling_synchronizer.hpp
         data_sources/fdv2/streaming_synchronizer.hpp
         data_sources/fdv2/fdv2_data_source.hpp
+        data_sources/fdv2/cache_initializer.hpp
         flag_manager/flag_store.hpp
         flag_manager/flag_updater.hpp
         bindings/c/sdk.cpp

libs/client-sdk/src/data_sources/data_source_update_sink.hpp

@@ -50,7 +50,7 @@ class IDataSourceUpdateSink {
      * instead.
      *
      * @param from_cache Whether the changeset was loaded from the local
-     * cache, in which case it is not written back to it.
+     * cache. Such data is not written back, nor treated as current.
      */
     virtual void Apply(Context const& context,
                        FlagChangeSet change_set,

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

@@ -0,0 +1,60 @@
+#include "cache_initializer.hpp"
+
+#include <utility>
+
+namespace launchdarkly::client_side::data_sources {
+
+static char const* const kIdentity = "FDv2 cache initializer";
+
+FDv2CacheInitializer::FDv2CacheInitializer(flag_manager::FlagPersistence* cache,
+                                           Context context,
+                                           Logger const& logger)
+    : cache_(cache), context_(std::move(context)), logger_(logger) {}
+
+async::Future<FDv2SourceResult> FDv2CacheInitializer::Run() {
+    auto data = cache_->ReadCached(context_);
+    if (!data) {
+        LD_LOG(logger_, LogLevel::kDebug)
+            << kIdentity << ": no cached data for this context";
+        // A miss leaves the data set unchanged so initialization can continue.
+        return async::MakeFuture(FDv2SourceResult{FDv2SourceResult::ChangeSet{
+            FlagChangeSet{data_model::ChangeSetType::kNone,
+                          {},
+                          data_model::Selector{}}}});
+    }
+
+    LD_LOG(logger_, LogLevel::kDebug)
+        << kIdentity << ": loaded " << data->size()
+        << " flags for this context";
+
+    FlagChangeSetData changes;
+    changes.reserve(data->size());
+    for (auto& [key, item] : *data) {
+        changes.push_back(FlagChange{key, std::move(item)});
+    }
+
+    return async::MakeFuture(FDv2SourceResult{FDv2SourceResult::ChangeSet{
+        FlagChangeSet{data_model::ChangeSetType::kFull, std::move(changes),
+                      data_model::Selector{}}}});
+}
+
+void FDv2CacheInitializer::Close() {
+    // Run() completes on the calling thread, so there is nothing to cancel.
+}
+
+std::string const& FDv2CacheInitializer::Identity() const {
+    static std::string const identity = kIdentity;
+    return identity;
+}
+
+FDv2CacheInitializerFactory::FDv2CacheInitializerFactory(
+    flag_manager::FlagPersistence* cache,
+    Context context,
+    Logger const& logger)
+    : cache_(cache), context_(std::move(context)), logger_(logger) {}
+
+std::unique_ptr<IFDv2Initializer> FDv2CacheInitializerFactory::Build() {
+    return std::make_unique<FDv2CacheInitializer>(cache_, context_, logger_);
+}
+
+}  // namespace launchdarkly::client_side::data_sources

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

@@ -0,0 +1,64 @@
+#pragma once
+
+#include "ifdv2_initializer.hpp"
+#include "ifdv2_initializer_factory.hpp"
+
+#include "../../flag_manager/flag_persistence.hpp"
+
+#include <launchdarkly/async/promise.hpp>
+#include <launchdarkly/context.hpp>
+#include <launchdarkly/logging/logger.hpp>
+
+#include <string>
+
+namespace launchdarkly::client_side::data_sources {
+
+/**
+ * Loads flag data the SDK persisted for this context on a previous run, so
+ * that evaluation can begin before the network answers.
+ */
+class FDv2CacheInitializer final : public IFDv2Initializer {
+   public:
+    /**
+     * @param cache Local cache to read. Non-owning. Must outlive this object.
+     * @param context The evaluation context to load data for.
+     * @param logger Receives diagnostic logging.
+     */
+    FDv2CacheInitializer(flag_manager::FlagPersistence* cache,
+                         Context context,
+                         Logger const& logger);
+
+    async::Future<FDv2SourceResult> Run() override;
+
+    void Close() override;
+
+    [[nodiscard]] std::string const& Identity() const override;
+
+   private:
+    flag_manager::FlagPersistence* const cache_;
+    Context const context_;
+    Logger logger_;
+};
+
+/**
+ * Builds fresh FDv2CacheInitializer instances on demand.
+ *
+ * Thread-safe: Build() may be called from any thread.
+ */
+class FDv2CacheInitializerFactory final : public IFDv2InitializerFactory {
+   public:
+    FDv2CacheInitializerFactory(flag_manager::FlagPersistence* cache,
+                                Context context,
+                                Logger const& logger);
+
+    std::unique_ptr<IFDv2Initializer> Build() override;
+
+    [[nodiscard]] bool IsFromCache() const override { return true; }
+
+   private:
+    flag_manager::FlagPersistence* const cache_;
+    Context const context_;
+    Logger logger_;
+};
+
+}  // namespace launchdarkly::client_side::data_sources

libs/client-sdk/src/flag_manager/context_index.cpp

@@ -23,13 +23,23 @@
     }
 }
 
+std::optional<std::chrono::time_point<std::chrono::system_clock>>
+ContextIndex::GetTimestamp(std::string const& id) const {
+    for (auto const& entry : index_) {
+        if (entry.id == id) {
+            return entry.timestamp;
+        }
+    }
+    return std::nullopt;
+}
+
 std::vector<std::string> ContextIndex::Prune(std::size_t maxContexts) {
     if (index_.size() <= maxContexts) {
         return {};
     }
 
     std::sort(index_.begin(), index_.end(),
               [](IndexEntry const& a, IndexEntry const& b) {
                   return a.timestamp > b.timestamp;
               });
 
@@ -60,7 +70,7 @@
     auto& top = json_value.emplace_object();
     auto arr = boost::json::array();
 
     for (auto& entry : index.Entries()) {
         auto obj = boost::json::object();
         obj.emplace("id", entry.id);
         obj.emplace("timestamp",
@@ -78,15 +88,15 @@
 
     auto index = ContextIndex::Index();
     if (json_value.is_object()) {
         auto arr = json_value.as_object().find("index");
         if (arr != json_value.as_object().end() && arr->value().is_array()) {
             for (auto& item : arr->value().as_array()) {
                 if (item.is_object()) {
                     auto& obj = item.as_object();
                     auto* id_iter = obj.find("id");
                     auto id = ValueAsOpt<std::string>(id_iter, obj.end());
 
                     auto* timestamp_iter = obj.find("timestamp");
                     auto timestamp =
                         ValueAsOpt<uint64_t>(timestamp_iter, obj.end());
 

libs/client-sdk/src/flag_manager/context_index.hpp

@@ -2,6 +2,7 @@
 
 #include <chrono>
 #include <mutex>
+#include <optional>
 #include <string>
 #include <vector>
 
@@ -20,6 +21,8 @@ namespace launchdarkly::client_side::flag_manager {
  * 1. a context identifier (hashed fully-qualified key) and
  * 2. timestamp when it was last accessed, to support an LRU
  * eviction pattern.
+ *
+ * Not thread-safe.
  */
 class ContextIndex {
    public:
@@ -52,6 +55,11 @@ class ContextIndex {
 
     [[nodiscard]] Index const& Entries() const;
 
+    /** Returns the timestamp recorded for the id, or nullopt if absent. */
+    [[nodiscard]] std::optional<
+        std::chrono::time_point<std::chrono::system_clock>>
+    GetTimestamp(std::string const& id) const;
+
     /**
      * Prune the index returning a list of the removed context keys
      *

libs/client-sdk/src/flag_manager/flag_manager.cpp

@@ -29,6 +29,10 @@ FlagStore const& FlagManager::Store() const {
     return flag_store_;
 }
 
+FlagPersistence& FlagManager::Cache() {
+    return persistence_updater_;
+}
+
 void FlagManager::LoadCache(Context const& context) {
     persistence_updater_.LoadCached(context);
 }

libs/client-sdk/src/flag_manager/flag_manager.hpp

@@ -8,6 +8,11 @@
 
 namespace launchdarkly::client_side::flag_manager {
 
+/**
+ * Owns the flag store and the update pipeline that feeds it.
+ *
+ * Thread-safe: each part it exposes is itself thread-safe.
+ */
 class FlagManager {
    public:
     FlagManager(std::string const& sdk_key,
@@ -18,6 +23,9 @@ class FlagManager {
     IFlagNotifier& Notifier();
     FlagStore const& Store() const;
 
+    /** Returns the local cache the SDK persists flag data to. */
+    FlagPersistence& Cache();
+
     void LoadCache(Context const& context);
 
    private:

libs/client-sdk/src/flag_manager/flag_persistence.cpp

@@ -4,6 +4,7 @@
 #include <launchdarkly/encoding/sha_256.hpp>
 
 #include <launchdarkly/detail/serialization/json_primitives.hpp>
+#include <launchdarkly/serialization/json_context.hpp>
 #include <launchdarkly/serialization/json_evaluation_result.hpp>
 #include <launchdarkly/serialization/json_item_descriptor.hpp>
 
@@ -59,25 +60,30 @@ void FlagPersistence::Apply(Context const& context,
         change_set.type != data_model::ChangeSetType::kNone;
     sink_.Apply(context, std::move(change_set), from_cache);
     if (from_cache) {
-        // Writing cached data back to the cache it came from would be a no-op.
+        // Writing cached data back to the cache would be a no-op, and it was
+        // never confirmed current by the service.
         return;
     }
+    // Both a payload and a confirmation that nothing changed mean the cache
+    // is up to date as of now.
+    RecordFreshness(context);
     if (changed_data) {
         StoreCache(PersistenceEncodeKey(context.CanonicalKey()));
     }
 }
 
-void FlagPersistence::LoadCached(Context const& context) {
+std::optional<std::unordered_map<std::string, ItemDescriptor>>
+FlagPersistence::ReadCached(Context const& context) {
     if (!persistence_ || !context.Valid()) {
-        return;
+        return std::nullopt;
     }
 
     std::lock_guard lock(persistence_mutex_);
     auto data = persistence_->Read(
         environment_namespace_, PersistenceEncodeKey(context.CanonicalKey()));
 
     if (!data) {
-        return;
+        return std::nullopt;
     }
 
     boost::system::error_code error_code;
@@ -86,24 +92,84 @@ void FlagPersistence::LoadCached(Context const& context) {
         LD_LOG(logger_, LogLevel::kError)
             << "Failed to parse flag data from persistence: "
             << error_code.message();
-        return;
+        return std::nullopt;
     }
 
     auto res = boost::json::value_to<tl::expected<
         std::optional<std::unordered_map<std::string, ItemDescriptor>>,
         JsonError>>(parsed);
     if (!res) {
         LD_LOG(logger_, LogLevel::kError)
-            << "Failed to parse flag data from persistence: "
-            << error_code.message();
-        return;
+            << "Failed to parse flag data from persistence";
+        return std::nullopt;
     }
 
     // If the map was null or omitted, treat it like an empty data set.
-    auto map =
-        res.value().value_or(std::unordered_map<std::string, ItemDescriptor>{});
+    return res.value().value_or(
+        std::unordered_map<std::string, ItemDescriptor>{});
+}
 
-    sink_.Init(context, std::move(map));
+void FlagPersistence::LoadCached(Context const& context) {
+    if (auto data = ReadCached(context)) {
+        sink_.Init(context, std::move(*data));
+    }
+}
+
+static bool HasPrefix(std::string const& value, std::string const& prefix) {
+    return value.compare(0, prefix.size(), prefix) == 0;
+}
+
+// Groups freshness records by the canonical key that flag data is stored under.
+static std::string FreshnessGroup(Context const& context) {
+    return PersistenceEncodeKey(context.CanonicalKey()) + ".";
+}
+
+// Identifies a context by everything it carries, not just its key. Changing
+// an attribute can change how flags evaluate.
+static std::string FreshnessId(Context const& context) {
+    return FreshnessGroup(context) +
+           PersistenceEncodeKey(
+               boost::json::serialize(boost::json::value_from(context)));
+}
+
+void FlagPersistence::RecordFreshness(Context const& context) {
+    if (!persistence_ || !context.Valid()) {
+        return;
+    }
+
+    std::lock_guard lock(persistence_mutex_);
+    auto index = ReadIndexAt(freshness_key_);
+
+    // Storing these flags replaced whatever shared the canonical key, so the
+    // other attribute sets in the group no longer describe stored data.
+    auto const group = FreshnessGroup(context);
+    ContextIndex::Index retained;
+    for (auto const& entry : index.Entries()) {
+        if (!HasPrefix(entry.id, group)) {
+            retained.push_back(entry);
+        }
+    }
+
+    ContextIndex regrouped{std::move(retained)};
+    regrouped.Notice(FreshnessId(context), time_stamper_());
+    regrouped.Prune(max_cached_contexts_);
+    persistence_->Set(
+        environment_namespace_, freshness_key_,
+        boost::json::serialize(boost::json::value_from(regrouped)));
+}
+
+std::optional<std::chrono::time_point<std::chrono::system_clock>>
+FlagPersistence::ReadFreshness(Context const& context) {
+    if (!persistence_ || !context.Valid()) {
+        return std::nullopt;
+    }
+
+    std::lock_guard lock(persistence_mutex_);
+    // Freshness must not outlive the flag data it describes.
+    if (!ReadCached(context)) {
+        return std::nullopt;
+    }
+    return ReadIndexAt(freshness_key_).GetTimestamp(FreshnessId(context));
 }
 
 void FlagPersistence::StoreCache(std::string const& context_id) {
@@ -112,7 +178,7 @@ void FlagPersistence::StoreCache(std::string const& context_id) {
     }
 
     std::lock_guard lock(persistence_mutex_);
-    auto index = GetIndex();
+    auto index = ReadIndexAt(index_key_);
     index.Notice(context_id, time_stamper_());
     auto pruned = index.Prune(max_cached_contexts_);
     for (auto& id : pruned) {
@@ -127,11 +193,9 @@ void FlagPersistence::StoreCache(std::string const& context_id) {
                       boost::json::serialize(v));
 }
 
-ContextIndex FlagPersistence::GetIndex() {
+ContextIndex FlagPersistence::ReadIndexAt(std::string const& key) {
     if (persistence_) {
-        std::lock_guard lock(persistence_mutex_);
-        auto index_data =
-            persistence_->Read(environment_namespace_, index_key_);
+        auto index_data = persistence_->Read(environment_namespace_, key);
 
         if (index_data) {
             boost::system::error_code error_code;