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,120 @@ use crate::{
         CommandParams, CommandResponse, EvaluateAllFlagsParams, EvaluateAllFlagsResponse,
         EvaluateFlagParams, EvaluateFlagResponse,
     },
-    CreateInstanceParams,
+    CreateInstanceParams, DataSynchronizerParams, 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.streaming_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.polling_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);
+        }
+    }
+
+    // Use the configured FDv1 fallback if present, otherwise derive one from the
+    // synchronizers. The fallback is always polling.
+    let fallback = match params.fdv1_fallback {
+        Some(fallback) => Some((
+            fallback
+                .base_uri
+                .or_else(|| derive_fallback_base_url(&synchronizers)),
+            fallback.poll_interval_ms,
+        )),
+        None => select_fallback_synchronizer(&synchronizers).map(|sync| {
+            if let Some(polling) = &sync.polling {
+                (polling.base_uri.clone(), polling.poll_interval_ms)
+            } else if let Some(streaming) = &sync.streaming {
+                (streaming.base_uri.clone(), None)
+            } else {
+                (None, None)
+            }
+        }),
+    };
+
+    if let Some((base_url, poll_interval)) = fallback {
+        if let Some(base_url) = base_url {
+            service_endpoints.polling_base_url(&base_url);
+        }
+        let mut fallback_builder = PollingDataSourceBuilder::<T>::new();
+        if let Some(interval) = poll_interval {
+            fallback_builder.poll_interval(Duration::from_millis(interval));
+        }
+        fallback_builder.transport(make_transport()?);
+        builder.fdv1_fallback(&fallback_builder);
+    }
+
+    Ok(builder)
+}
+
+/// Selects the synchronizer to derive an FDv1 fallback from: the first polling
+/// synchronizer, otherwise the first synchronizer.
+fn select_fallback_synchronizer(
+    synchronizers: &[DataSynchronizerParams],
+) -> Option<&DataSynchronizerParams> {
+    synchronizers
+        .iter()
+        .find(|sync| sync.polling.is_some())
+        .or_else(|| synchronizers.first())
+}
+
+/// Derives an FDv1 fallback base URL from the synchronizers.
+fn derive_fallback_base_url(synchronizers: &[DataSynchronizerParams]) -> Option<String> {
+    let sync = select_fallback_synchronizer(synchronizers)?;
+    if let Some(polling) = &sync.polling {
+        polling.base_uri.clone()
+    } else if let Some(streaming) = &sync.streaming {
+        streaming.base_uri.clone()
+    } else {
+        None
+    }
+}
+
 pub struct ClientEntity {
     client: Arc<Client>,
 }
@@ -45,15 +158,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 +202,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 +218,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 +230,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 +264,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,38 @@
+# 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
+
+# Rust honors the TTL-based FDv1 fallback directive (SDK-2527); this scenario
+# tests the terminal semantics instead.
+streaming/fdv2/FDv1 fallback directive/directive without FDv1 fallback configured halts the data system
+
+# Rust exposes no FDv2 payload filter API (SDK-2575).
+streaming/requests/URL path is computed correctly/environment_filter_key="encoding_not_necessary"/base URI has no trailing slash/GET
+streaming/requests/URL path is computed correctly/environment_filter_key="encoding_not_necessary"/base URI has a trailing slash/GET
+polling/requests/URL path is computed correctly/environment_filter_key="encoding_not_necessary"/base URI has no trailing slash/GET
+polling/requests/URL path is computed correctly/environment_filter_key="encoding_not_necessary"/base URI has a trailing slash/GET
+
+# Rust does not escape `/`- or `~`-prefixed redacted attribute names as attribute references in events (SDK-2923).
+events/context properties/single-kind, allAttributesPrivate, slash-prefixed attribute name/debug event
+events/context properties/single-kind, allAttributesPrivate, slash-prefixed attribute name/identify event
+events/context properties/single-kind, allAttributesPrivate, slash-prefixed attribute name/index event from custom event
+events/context properties/single-kind, allAttributesPrivate, slash-prefixed attribute name/index event from evaluation
+events/feature events/multi-kind with anonymous context redacts attributes appropriately/type: any
+events/feature events/multi-kind with anonymous context redacts attributes appropriately/type: bool
+events/feature events/multi-kind with anonymous context redacts attributes appropriately/type: double
+events/feature events/multi-kind with anonymous context redacts attributes appropriately/type: int
+events/feature events/multi-kind with anonymous context redacts attributes appropriately/type: string
+events/feature events/single-kind anonymous context redacts all attributes/type: any
+events/feature events/single-kind anonymous context redacts all attributes/type: bool
+events/feature events/single-kind anonymous context redacts all attributes/type: double
+events/feature events/single-kind anonymous context redacts all attributes/type: int
+events/feature events/single-kind anonymous context redacts all attributes/type: string