feat: Drive the client FDv2 data system from the contract tests

.github/workflows/client.yml

@@ -34,6 +34,31 @@ jobs:
           test_service_port: ${{ env.TEST_SERVICE_PORT }}
           token: ${{ secrets.GITHUB_TOKEN }}
 
+  contract-tests-fdv2:
+    runs-on: ubuntu-22.04
+    env:
+      TEST_SERVICE_PORT: 8123
+      TEST_SERVICE_BINARY: ./build/contract-tests/client-contract-tests/client-tests
+      SUPPRESSION_FILE: contract-tests/client-contract-tests/test-suppressions-fdv2.txt
+    steps:
+      # https://github.com/actions/checkout/releases/tag/v7.0.1
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1
+      - uses: ./.github/actions/ci
+        with:
+          cmake_target: client-tests
+          run_tests: false
+      - name: 'Launch test service as background task'
+        run: $TEST_SERVICE_BINARY $TEST_SERVICE_PORT 2>&1 &
+      # https://github.com/launchdarkly/gh-actions/releases/tag/contract-tests-v1.1.0
+      - uses: launchdarkly/gh-actions/actions/contract-tests@2715574e04448246bc529a23a81766491bbc4aae
+        with:
+          test_service_port: ${{ env.TEST_SERVICE_PORT }}
+          token: ${{ secrets.GITHUB_TOKEN }}
+          version: v3
+          branch: v3
+          debug_logging: 'true'
+          extra_params: -skip-from=${{ env.SUPPRESSION_FILE }}
+
   contract-tests-curl:
     runs-on: ubuntu-22.04
     env:
@@ -54,6 +79,33 @@ jobs:
           # Inform the test harness of test service's port.
           test_service_port: ${{ env.TEST_SERVICE_PORT }}
           token: ${{ secrets.GITHUB_TOKEN }}
+
+  contract-tests-fdv2-curl:
+    runs-on: ubuntu-22.04
+    env:
+      TEST_SERVICE_PORT: 8123
+      TEST_SERVICE_BINARY: ./build/contract-tests/client-contract-tests/client-tests
+      SUPPRESSION_FILE: contract-tests/client-contract-tests/test-suppressions-fdv2.txt
+    steps:
+      # https://github.com/actions/checkout/releases/tag/v7.0.1
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1
+      - uses: ./.github/actions/ci
+        with:
+          cmake_target: client-tests
+          run_tests: false
+          use_curl: true
+      - name: 'Launch test service as background task'
+        run: $TEST_SERVICE_BINARY $TEST_SERVICE_PORT 2>&1 &
+      # https://github.com/launchdarkly/gh-actions/releases/tag/contract-tests-v1.1.0
+      - uses: launchdarkly/gh-actions/actions/contract-tests@2715574e04448246bc529a23a81766491bbc4aae
+        with:
+          test_service_port: ${{ env.TEST_SERVICE_PORT }}
+          token: ${{ secrets.GITHUB_TOKEN }}
+          version: v3
+          branch: v3
+          debug_logging: 'true'
+          extra_params: -skip-from=${{ env.SUPPRESSION_FILE }}
+
   build-test:
     runs-on: ubuntu-22.04
     steps:

contract-tests/client-contract-tests/src/entity_manager.cpp

@@ -1,4 +1,4 @@
 #include "entity_manager.hpp"
 
 #include <launchdarkly/config/client.hpp>
 #include <launchdarkly/context_builder.hpp>
@@ -6,6 +6,10 @@
 
 #include <boost/json.hpp>
 
+#include <chrono>
+#include <optional>
+#include <utility>
+
 using launchdarkly::LogLevel;
 using namespace launchdarkly::client_side;
 
@@ -17,13 +21,147 @@
       logger_{logger} {}
 
 static tl::expected<launchdarkly::Context, launchdarkly::JsonError>
 ParseContext(nlohmann::json value) {
     auto boost_json_val = boost::json::parse(value.dump());
     return boost::json::value_to<
         tl::expected<launchdarkly::Context, launchdarkly::JsonError>>(
         boost_json_val);
 }
 
+static std::chrono::seconds ToSeconds(uint64_t milliseconds) {
+    return std::chrono::duration_cast<std::chrono::seconds>(
+        std::chrono::milliseconds(milliseconds));
+}
+
+// The harness names modes the way the configuration API does. Desktop
+// provides three of them. The mobile- and browser-only modes are not
+// configurable here.
+static std::optional<ConnectionMode> ParseConnectionMode(
+    std::string const& name) {
+    if (name == "streaming") {
+        return ConnectionMode::kStreaming;
+    }
+    if (name == "polling") {
+        return ConnectionMode::kPolling;
+    }
+    if (name == "offline") {
+        return ConnectionMode::kOffline;
+    }
+    return std::nullopt;
+}
+
+// Builds exactly the pipeline the harness asked for. No cache initializer is
+// added, so the requests the SDK makes are the ones the harness expects.
+static FDv2Builder::Mode BuildMode(
+    ConfigModeDefinitionParams const& definition,
+    std::optional<ConfigPollingParams> const& fdv1_fallback) {
+    FDv2Builder::Mode mode;
+
+    if (definition.initializers) {
+        for (auto const& initializer : *definition.initializers) {
+            if (!initializer.polling) {
+                continue;
+            }
+            auto polling = FDv2Builder::Polling();
+            if (initializer.polling->baseUri) {
+                polling.BaseUrl(*initializer.polling->baseUri);
+            }
+            if (initializer.polling->pollIntervalMs) {
+                polling.PollInterval(
+                    ToSeconds(*initializer.polling->pollIntervalMs));
+            }
+            mode.Initializer(std::move(polling));
+        }
+    }
+
+    if (definition.synchronizers) {
+        for (auto const& synchronizer : *definition.synchronizers) {
+            if (synchronizer.streaming) {
+                auto streaming = FDv2Builder::Streaming();
+                if (synchronizer.streaming->baseUri) {
+                    streaming.BaseUrl(*synchronizer.streaming->baseUri);
+                }
+                if (synchronizer.streaming->initialRetryDelayMs) {
+                    streaming.InitialReconnectDelay(std::chrono::milliseconds(
+                        *synchronizer.streaming->initialRetryDelayMs));
+                }
+                mode.Synchronizer(std::move(streaming));
+            } else if (synchronizer.polling) {
+                auto polling = FDv2Builder::Polling();
+                if (synchronizer.polling->baseUri) {
+                    polling.BaseUrl(*synchronizer.polling->baseUri);
+                }
+                if (synchronizer.polling->pollIntervalMs) {
+                    polling.PollInterval(
+                        ToSeconds(*synchronizer.polling->pollIntervalMs));
+                }
+                mode.Synchronizer(std::move(polling));
+            }
+        }
+    }
+
+    if (fdv1_fallback) {
+        auto fallback = FDv2Builder::FDv1Fallback();
+        if (fdv1_fallback->baseUri) {
+            fallback.BaseUrl(*fdv1_fallback->baseUri);
+        }
+        if (fdv1_fallback->pollIntervalMs) {
+            fallback.PollInterval(ToSeconds(*fdv1_fallback->pollIntervalMs));
+        }
+        mode.FallbackToFDv1(std::move(fallback));
+    } else {
+        mode.DisableFDv1Fallback();
+    }
+
+    return mode;
+}
+
+static bool HasPipelines(ConfigDataSystemParams const& cfg) {
+    return (cfg.initializers && !cfg.initializers->empty()) ||
+           (cfg.synchronizers && !cfg.synchronizers->empty());
+}
+
+static FDv2Builder BuildFDv2(ConfigDataSystemParams const& cfg) {
+    FDv2Builder fdv2;
+
+    if (cfg.useDefaultDataSystem.value_or(false)) {
+        return fdv2;
+    }
+
+    if (cfg.connectionModeConfig) {
+        auto const& modes = *cfg.connectionModeConfig;
+        if (modes.initialConnectionMode) {
+            if (auto const mode =
+                    ParseConnectionMode(*modes.initialConnectionMode)) {
+                fdv2.InitialMode(*mode);
+            }
+        }
+        if (modes.customConnectionModes) {
+            for (auto const& [name, definition] :
+                 *modes.customConnectionModes) {
+                if (auto const mode = ParseConnectionMode(name)) {
+                    fdv2.CustomizeMode(*mode,
+                                       BuildMode(definition, cfg.fdv1Fallback));
+                }
+            }
+        }
+        return fdv2;
+    }
+
+    // A pipeline given without a mode wrapper describes the mode the SDK
+    // starts in.
+    if (HasPipelines(cfg)) {
+        ConfigModeDefinitionParams top_level;
+        top_level.initializers = cfg.initializers;
+        top_level.synchronizers = cfg.synchronizers;
+        fdv2.InitialMode(ConnectionMode::kStreaming);
+        fdv2.CustomizeMode(ConnectionMode::kStreaming,
+                           BuildMode(top_level, cfg.fdv1Fallback));
+    }
+
+    return fdv2;
+}
+
 std::optional<std::string> EntityManager::create(ConfigParams const& in) {
     std::string id = std::to_string(counter_++);
 
@@ -126,6 +264,12 @@
         datasource.UseReport(*in.clientSide->useReport);
     }
 
+    // A data system configuration selects FDv2, which supersedes the
+    // streaming and polling methods above.
+    if (in.dataSystem) {
+        datasource.Method(BuildFDv2(*in.dataSystem));
+    }
+
     if (in.tags) {
         if (in.tags->applicationId) {
             config_builder.AppInfo().Identifier(*in.tags->applicationId);

contract-tests/client-contract-tests/src/main.cpp

@@ -1,4 +1,4 @@
 #include "server.hpp"
 
 #include <launchdarkly/logging/console_backend.hpp>
 
@@ -18,7 +18,7 @@
 using launchdarkly::LogLevel;
 
 int main(int argc, char* argv[]) {
     launchdarkly::Logger logger{
         std::make_unique<ConsoleBackend>("client-contract-tests")};
 
     std::string const default_port = "8123";
@@ -31,8 +31,8 @@
     try {
         net::io_context ioc{1};
 
         auto p = boost::lexical_cast<unsigned short>(port);
         server srv(ioc, "0.0.0.0", p, logger);
 
         srv.add_capability("client-side");
         srv.add_capability("mobile");
@@ -49,6 +49,8 @@
         srv.add_capability("client-prereq-events");
         srv.add_capability("client-prereq-cycle-detection");
         srv.add_capability("wrapper");
+        // The FDv2 data system honors the service's FDv1 fallback directive.
+        srv.add_capability("fdv1-fallback");
         // Proxies are supported only with CURL networking.
 #ifdef LD_CURL_NETWORKING
         srv.add_capability("http-proxy");

contract-tests/client-contract-tests/test-suppressions-fdv2.txt

@@ -0,0 +1,19 @@
+# In the C++ client a fallback directive schedules a return to FDv2 rather
+# than ending the data system.
+# These two scenarios assert the older terminal semantics, and the server SDK
+# suppresses the same pair for the same reason. Tracked in SDK-2527.
+#
+# Both are currently skipped for other reasons anyway. One needs the
+# "client-event-source-http-errors" capability and the other needs
+# -enable-long-running-tests. These entries are a guard for when either
+# changes rather than something the suite exercises today.
+streaming/fdv2/FDv1 fallback directive/directive without FDv1 fallback configured halts the data system
+streaming/fdv2/FDv1 fallback directive/directed fallback is terminal and does not revisit FDv2 sources
+
+# The event processor gates SendAsync on an inbox counter sized at the events
+# capacity, so a burst that outruns the io_context drops events before they
+# reach the summarizer. This test sends a full buffer of custom events and then
+# evaluates a flag, and under FDv2 the busier startup makes the evaluation lose
+# that race, so no summary event is produced. The outbox itself already keeps
+# the summary when it overflows. The inbox is the gap.
+events/event capacity/summary event is still included even if buffer was full

contract-tests/data-model/include/data_model/data_model.hpp

@@ -1,9 +1,10 @@
 #pragma once
 
+#include <map>
 #include <optional>
 #include <string>
 #include <unordered_map>
 #include "nlohmann/json.hpp"
 
 namespace nlohmann {
 template <typename T>
@@ -65,7 +66,7 @@
                                                 pollIntervalMs,
                                                 filter);
 
 struct ConfigEventParams {
     std::optional<std::string> baseUri;
     std::optional<uint32_t> capacity;
     std::optional<bool> enableDiagnostics;
@@ -93,7 +94,7 @@
                                                 polling,
                                                 events);
 
 struct ConfigClientSideParams {
     nlohmann::json initialContext;
     std::optional<bool> evaluationReasons;
     std::optional<bool> useReport;
@@ -136,7 +137,7 @@
                                                 data,
                                                 errors);
 
 struct ConfigHooksParams {
     std::vector<ConfigHookInstance> hooks;
 };
 
@@ -201,20 +202,46 @@
 NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE_WITH_DEFAULT(ConfigDataInitializerParams,
                                                 polling);
 
+// One connection mode's pipeline, for client-side SDKs.
+struct ConfigModeDefinitionParams {
+    std::optional<std::vector<ConfigDataInitializerParams>> initializers;
+    std::optional<std::vector<ConfigDataSynchronizerParams>> synchronizers;
+};
+
+NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE_WITH_DEFAULT(ConfigModeDefinitionParams,
+                                                initializers,
+                                                synchronizers);
+
+// Which connection mode a client-side SDK starts in, and how any of them are
+// customized.
+struct ConfigConnectionModeParams {
+    std::optional<std::string> initialConnectionMode;
+    std::optional<std::map<std::string, ConfigModeDefinitionParams>>
+        customConnectionModes;
+};
+
+NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE_WITH_DEFAULT(ConfigConnectionModeParams,
+                                                initialConnectionMode,
+                                                customConnectionModes);
+
 struct ConfigDataSystemParams {
     std::optional<std::vector<ConfigDataInitializerParams>> initializers;
     std::optional<std::vector<ConfigDataSynchronizerParams>> synchronizers;
     std::optional<ConfigPollingParams> fdv1Fallback;
     std::optional<std::string> payloadFilter;
     std::optional<ConfigDataSystemStore> store;
+    std::optional<bool> useDefaultDataSystem;
+    std::optional<ConfigConnectionModeParams> connectionModeConfig;
 };
 
 NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE_WITH_DEFAULT(ConfigDataSystemParams,
                                                 initializers,
                                                 synchronizers,
                                                 fdv1Fallback,
                                                 payloadFilter,
-                                                store);
+                                                store,
+                                                useDefaultDataSystem,
+                                                connectionModeConfig);
 
 struct ConfigBigSegmentsParams {
     std::string callbackUri;
@@ -268,7 +295,7 @@
                                                 persistentDataStore,
                                                 bigSegments);
 
 struct ContextSingleParams {
     std::optional<std::string> kind;
     std::string key;
     std::optional<std::string> name;
@@ -363,7 +390,7 @@
                                                 defaultValue,
                                                 detail);
 
 struct EvaluateFlagResponse {
     nlohmann::json value;
     std::optional<uint32_t> variationIndex;
     std::optional<nlohmann::json> reason;