test: Run the FDv2 contract tests

.github/actions/contract-tests/action.yml

@@ -19,8 +19,20 @@ runs:
       shell: bash
       run: CARGO_FLAGS="${{ inputs.cargo-flags }}" make start-contract-test-service-bg
 
-    - uses: launchdarkly/gh-actions/actions/contract-tests@contract-tests-v1.0.2
+    - name: Run FDv1 contract tests
+      uses: launchdarkly/gh-actions/actions/contract-tests@contract-tests-v1.3.0
       with:
         test_service_port: 8000
         token: ${{ inputs.token }}
         extra_params: "-skip-from ./contract-tests/testharness-suppressions.txt"
+        enable_persistence_tests: false
+        stop_service: false
+
+    - name: Run FDv2 contract tests
+      uses: launchdarkly/gh-actions/actions/contract-tests@contract-tests-v1.3.0
+      with:
+        test_service_port: 8000
+        token: ${{ inputs.token }}
+        version: v3
+        extra_params: "-skip-from ./contract-tests/testharness-suppressions-fdv2.txt"
+        enable_persistence_tests: false

Makefile

@@ -17,4 +17,10 @@ run-contract-tests:
 
 contract-tests: build-contract-tests start-contract-test-service-bg run-contract-tests
 
-.PHONY: build-contract-tests start-contract-test-service run-contract-tests contract-tests
+run-contract-tests-fdv2:
+	@curl -s https://raw.githubusercontent.com/launchdarkly/sdk-test-harness/main/downloader/run.sh \
+      | VERSION=v3 PARAMS="-url http://localhost:8000 -debug -stop-service-at-end -skip-from ./contract-tests/testharness-suppressions-fdv2.txt $(TEST_HARNESS_PARAMS)" sh
+
+contract-tests-fdv2: build-contract-tests start-contract-test-service-bg run-contract-tests-fdv2
+
+.PHONY: build-contract-tests start-contract-test-service run-contract-tests contract-tests run-contract-tests-fdv2 contract-tests-fdv2

contract-tests/src/client_entity.rs

@@ -9,10 +9,12 @@ const DEFAULT_POLLING_BASE_URL: &str = "https://sdk.launchdarkly.com";
 const DEFAULT_STREAM_BASE_URL: &str = "https://stream.launchdarkly.com";
 const DEFAULT_EVENTS_BASE_URL: &str = "https://events.launchdarkly.com";
 
+use launchdarkly_sdk_transport::HttpTransport;
 use launchdarkly_server_sdk::{
-    ApplicationInfo, BuildError, Client, ConfigBuilder, Detail, EventProcessorBuilder,
-    FlagDetailConfig, FlagFilter, FlagValue, NullEventProcessorBuilder, PollingDataSourceBuilder,
-    ServiceEndpointsBuilder, StreamingDataSourceBuilder,
+    ApplicationInfo, BuildError, Client, ConfigBuilder, DataSystemBuilder, Detail,
+    EventProcessorBuilder, FDv2PollingBuilder, FDv2StreamingBuilder, FlagDetailConfig, FlagFilter,
+    FlagValue, NullEventProcessorBuilder, PollingDataSourceBuilder, ServiceEndpointsBuilder,
+    StreamingDataSourceBuilder,
 };
 
 #[cfg(any(feature = "crypto-aws-lc-rs", feature = "crypto-openssl"))]
@@ -27,9 +29,77 @@ use crate::{
         CommandParams, CommandResponse, EvaluateAllFlagsParams, EvaluateAllFlagsResponse,
         EvaluateFlagParams, EvaluateFlagResponse,
     },
-    CreateInstanceParams,
+    CreateInstanceParams, DataSystemParams,
 };
 
+/// Builds an FDv2 data system from the contract test's data system params. The
+/// FDv1 fallback is always polling and reads its base URL from the polling
+/// service endpoint, so it is set here alongside the fallback source.
+fn build_fdv2_data_system<T, F>(
+    params: DataSystemParams,
+    make_transport: F,
+    service_endpoints: &mut ServiceEndpointsBuilder,
+) -> Result<DataSystemBuilder, BuildError>
+where
+    T: HttpTransport + Clone + Send + Sync + 'static,
+    F: Fn() -> Result<T, BuildError>,
+{
+    let mut builder = DataSystemBuilder::custom();
+
+    let synchronizers = params.synchronizers.unwrap_or_default();
+    for sync in &synchronizers {
+        if let Some(streaming) = &sync.streaming {
+            let mut source = FDv2StreamingBuilder::<T>::new();
+            if let Some(base_uri) = &streaming.base_uri {
+                source.base_url(base_uri);
+            }
+            if let Some(delay) = streaming.initial_retry_delay_ms {
+                source.initial_reconnect_delay(Duration::from_millis(delay));
+            }
+            source.transport(make_transport()?);
+            builder.synchronizer(source);
+        } else if let Some(polling) = &sync.polling {
+            let mut source = FDv2PollingBuilder::<T>::new();
+            if let Some(base_uri) = &polling.base_uri {
+                source.base_url(base_uri);
+            }
+            if let Some(interval) = polling.poll_interval_ms {
+                source.poll_interval(Duration::from_millis(interval));
+            }
+            source.transport(make_transport()?);
+            builder.synchronizer(source);
+        }
+    }
+
+    for init in params.initializers.unwrap_or_default() {
+        if let Some(polling) = &init.polling {
+            let mut source = FDv2PollingBuilder::<T>::new();
+            if let Some(base_uri) = &polling.base_uri {
+                source.base_url(base_uri);
+            }
+            if let Some(interval) = polling.poll_interval_ms {
+                source.poll_interval(Duration::from_millis(interval));
+            }
+            source.transport(make_transport()?);
+            builder.initializer(source);
+        }
+    }
+
+    if let Some(fallback) = params.fdv1_fallback {
+        if let Some(base_url) = fallback.base_uri {
+            service_endpoints.polling_base_url(&base_url);
+        }
+        let mut fallback_builder = PollingDataSourceBuilder::<T>::new();
+        if let Some(interval) = fallback.poll_interval_ms {
+            fallback_builder.poll_interval(Duration::from_millis(interval));
+        }
+        fallback_builder.transport(make_transport()?);
+        builder.fdv1_fallback(&fallback_builder);
+    }
+
+    Ok(builder)
+}
+
 pub struct ClientEntity {
     client: Arc<Client>,
 }
@@ -45,15 +115,17 @@ impl ClientEntity {
             .unwrap_or_default()
             .http_proxy
             .unwrap_or_default();
-        let mut transport_builder = launchdarkly_sdk_transport::HyperTransport::builder();
-        if !proxy.is_empty() {
-            transport_builder = transport_builder.proxy_url(proxy.clone());
-        }
-
-        // Create fresh transports for this client to avoid shared connection pool issues
-        let transport = transport_builder
-            .build_with_connector(connector.clone())
-            .map_err(|e| BuildError::InvalidConfig(e.to_string()))?;
+        // Build a fresh transport per component, as the SDK normally does. Only the
+        // connector under test is shared across them.
+        let make_transport = || {
+            let mut builder = launchdarkly_sdk_transport::HyperTransport::builder();
+            if !proxy.is_empty() {
+                builder = builder.proxy_url(proxy.clone());
+            }
+            builder
+                .build_with_connector(connector.clone())
+                .map_err(|e| BuildError::InvalidConfig(e.to_string()))
+        };
         let mut config_builder =
             ConfigBuilder::new(&create_instance_params.configuration.credential);
 
@@ -87,7 +159,14 @@ impl ClientEntity {
             }
         }
 
-        if let Some(streaming) = create_instance_params.configuration.streaming {
+        if let Some(data_system) = create_instance_params.configuration.data_system {
+            let data_system_builder = build_fdv2_data_system(
+                data_system,
+                make_transport,
+                &mut service_endpoints_builder,
+            )?;
+            config_builder = config_builder.data_system(&data_system_builder);
+        } else if let Some(streaming) = create_instance_params.configuration.streaming {
             if let Some(base_uri) = streaming.base_uri {
                 service_endpoints_builder.streaming_base_url(&base_uri);
             }
@@ -96,7 +175,7 @@ impl ClientEntity {
             if let Some(delay) = streaming.initial_retry_delay_ms {
                 streaming_builder.initial_reconnect_delay(Duration::from_millis(delay));
             }
-            streaming_builder.transport(transport.clone());
+            streaming_builder.transport(make_transport()?);
 
             config_builder = config_builder.data_source(&streaming_builder);
         } else if let Some(polling) = create_instance_params.configuration.polling {
@@ -108,15 +187,15 @@ impl ClientEntity {
             if let Some(delay) = polling.poll_interval_ms {
                 polling_builder.poll_interval(Duration::from_millis(delay));
             }
-            polling_builder.transport(transport.clone());
+            polling_builder.transport(make_transport()?);
 
             config_builder = config_builder.data_source(&polling_builder);
         } else {
             // If we didn't specify streaming or polling, we fall back to basic streaming. The only
             // customization we provide is the transport to support testing multiple
             // transport implementations.
             let mut streaming_builder = StreamingDataSourceBuilder::new();
-            streaming_builder.transport(transport.clone());
+            streaming_builder.transport(make_transport()?);
             config_builder = config_builder.data_source(&streaming_builder);
         }
 
@@ -142,7 +221,7 @@ impl ClientEntity {
             if let Some(attributes) = events.global_private_attributes {
                 processor_builder.private_attributes(attributes);
             }
-            processor_builder.transport(transport);
+            processor_builder.transport(make_transport()?);
             processor_builder.omit_anonymous_contexts(events.omit_anonymous_contexts);
 
             config_builder.event_processor(&processor_builder)

contract-tests/src/main.rs

@@ -68,6 +68,27 @@ pub struct ServiceEndpointParameters {
     pub events: Option<String>,
 }
 
+#[derive(Deserialize, Debug)]
+#[serde(rename_all = "camelCase")]
+pub struct DataInitializerParams {
+    pub polling: Option<PollingParameters>,
+}
+
+#[derive(Deserialize, Debug)]
+#[serde(rename_all = "camelCase")]
+pub struct DataSynchronizerParams {
+    pub streaming: Option<StreamingParameters>,
+    pub polling: Option<PollingParameters>,
+}
+
+#[derive(Deserialize, Debug)]
+#[serde(rename_all = "camelCase")]
+pub struct DataSystemParams {
+    pub initializers: Option<Vec<DataInitializerParams>>,
+    pub synchronizers: Option<Vec<DataSynchronizerParams>>,
+    pub fdv1_fallback: Option<PollingParameters>,
+}
+
 #[derive(Deserialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct Configuration {
@@ -89,6 +110,8 @@ pub struct Configuration {
     pub tags: Option<TagParams>,
 
     pub service_endpoints: Option<ServiceEndpointParameters>,
+
+    pub data_system: Option<DataSystemParams>,
 }
 
 #[derive(Deserialize, Debug)]
@@ -123,6 +146,7 @@ async fn status() -> impl Responder {
             "event-gzip".to_string(),
             "optional-event-gzip".to_string(),
             "instance-id".to_string(),
+            "fdv1-fallback".to_string(),
         ],
     })
 }

contract-tests/testharness-suppressions-fdv2.txt

@@ -0,0 +1,12 @@
+# Suppressions for the FDv2 (v3) harness run. The v3 harness runs both the FDv1
+# and FDv2 suites, so this file carries the FDv1 suppressions as well.
+
+# FDv1 suppressions (also run under the v3 harness).
+evaluation/all flags state/client not ready
+evaluation/client not ready
+streaming/validation/drop and reconnect if stream event has malformed JSON/delete event
+streaming/validation/drop and reconnect if stream event has malformed JSON/patch event
+streaming/validation/drop and reconnect if stream event has malformed JSON/put event
+streaming/validation/drop and reconnect if stream event has well-formed JSON not matching schema/delete event
+streaming/validation/drop and reconnect if stream event has well-formed JSON not matching schema/patch event
+streaming/validation/drop and reconnect if stream event has well-formed JSON not matching schema/put event

launchdarkly-server-sdk/Cargo.toml

@@ -27,7 +27,7 @@ log = "0.4.14"
 lru = { version = "0.16.3", default-features = false }
 # Pulled without default features so that the float-roundtrip feature below is what decides
 # whether the evaluation engine parses floats the way Go does.
-launchdarkly-server-sdk-evaluation = { version = "2.2.0", default-features = false }
+launchdarkly-server-sdk-evaluation = { version = "2.2.1", default-features = false }
 serde = { version = "1.0.132", features = ["derive"] }
 serde_json = { version = "1.0.73" }
 thiserror = "2.0"