fix: honor server-directed FDv1 Fallback Directive in initializer phase

Previously the X-LD-FD-Fallback response header was honored only in
the synchronizer phase, and in the synchronizer phase a payload
arriving alongside the directive was discarded because the streaming
processor halted before applying it. Per the updated FDv2 Data System
spec, the directive must be honored in both initializer and
synchronizer phases, must take precedence over the section 1.2
failover algorithm, and must let any accompanying payload be applied
before the SDK switches terminally to the FDv1 Fallback Synchronizer.

Carry the signal explicitly on the public initializer/synchronizer
result types -- Basis.fallback_to_fdv1 and Update.fallback_to_fdv1 --
so callers cannot silently drop it. Update.revert_to_fdv1 is renamed
to Update.fallback_to_fdv1 to match the spec terminology.

Author: keelerm84

ldclient/impl/datasourcev2/polling.py

@@ -139,7 +139,7 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
                         yield Update(
                             state=DataSourceState.OFF,
                             error=error_info,
-                            revert_to_fdv1=True,
+                            fallback_to_fdv1=True,
                             environment_id=envid,
                         )
                         break
@@ -168,6 +168,17 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
                     message=result.error,
                 )
 
+                # Even a non-HTTP error (e.g. malformed JSON) can carry the fallback
+                # header. If so, halt rather than retrying the FDv2 endpoint.
+                if fallback:
+                    yield Update(
+                        state=DataSourceState.OFF,
+                        error=error_info,
+                        fallback_to_fdv1=True,
+                        environment_id=envid,
+                    )
+                    break
+
                 yield Update(
                     state=DataSourceState.INTERRUPTED,
                     error=error_info,
@@ -179,7 +190,7 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
                     state=DataSourceState.VALID,
                     change_set=change_set,
                     environment_id=headers.get(_LD_ENVID_HEADER),
-                    revert_to_fdv1=headers.get(_LD_FD_FALLBACK_HEADER) == 'true'
+                    fallback_to_fdv1=headers.get(_LD_FD_FALLBACK_HEADER) == 'true'
                 )
 
             if self._interrupt_event.wait(self._poll_interval):
@@ -204,13 +215,18 @@ def _poll(self, ss: SelectorStore) -> BasisResult:
                     if is_http_error_recoverable(status_code):
                         log.warning(http_error_message_result)
 
+                    # Forward any response headers so callers (e.g. FDv2 datasystem)
+                    # can read the X-LD-FD-Fallback directive even on error.
                     return _Fail(
-                        error=http_error_message_result, exception=result.exception
+                        error=http_error_message_result,
+                        exception=result.exception,
+                        headers=result.headers,
                     )
 
                 return _Fail(
                     error=result.error or "Failed to request payload",
                     exception=result.exception,
+                    headers=result.headers,
                 )
 
             (change_set, headers) = result.value
@@ -223,6 +239,7 @@ def _poll(self, ss: SelectorStore) -> BasisResult:
                 change_set=change_set,
                 persist=change_set.selector.is_defined(),
                 environment_id=env_id,
+                fallback_to_fdv1=headers.get(_LD_FD_FALLBACK_HEADER) == 'true',
             )
 
             return _Success(value=basis)

ldclient/impl/datasourcev2/streaming.py

@@ -166,6 +166,11 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
         self._connection_attempt_start_time = time()
 
         envid = None
+        # fallback_requested is set when a Start action carries
+        # X-LD-FD-Fallback: true. We finish applying the current payload
+        # before halting, so consumers can serve the server-provided data
+        # while FDv1 takes over.
+        fallback_requested = False
         for action in self._sse.all:
             if isinstance(action, Fault):
                 # If the SSE client detects the stream has closed, then it will
@@ -186,17 +191,9 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
                 continue
 
             if isinstance(action, Start) and action.headers is not None:
-                fallback = action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true'
                 envid = action.headers.get(_LD_ENVID_HEADER, envid)
-
-                if fallback:
-                    self._record_stream_init(True)
-                    yield Update(
-                        state=DataSourceState.OFF,
-                        revert_to_fdv1=True,
-                        environment_id=envid,
-                    )
-                    break
+                if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true':
+                    fallback_requested = True
 
             if not isinstance(action, Event):
                 continue
@@ -206,6 +203,18 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
                 if update is not None:
                     self._record_stream_init(False)
                     self._connection_attempt_start_time = None
+                    if fallback_requested:
+                        # Decorate the completed update with the fallback signal,
+                        # then halt — the consumer will switch to FDv1.
+                        update = Update(
+                            state=update.state,
+                            change_set=update.change_set,
+                            error=update.error,
+                            fallback_to_fdv1=True,
+                            environment_id=update.environment_id,
+                        )
+                        yield update
+                        break
                     yield update
             except json.decoder.JSONDecodeError as e:
                 log.info(
@@ -229,7 +238,7 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
                     error=DataSourceErrorInfo(
                         DataSourceErrorKind.UNKNOWN, 0, time(), str(e)
                     ),
-                    revert_to_fdv1=False,
+                    fallback_to_fdv1=False,
                     environment_id=envid,
                 )
 
@@ -353,7 +362,7 @@ def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optiona
                 error=DataSourceErrorInfo(
                     DataSourceErrorKind.INVALID_DATA, 0, time(), str(error)
                 ),
-                revert_to_fdv1=False,
+                fallback_to_fdv1=False,
                 environment_id=envid,
             )
             return (update, True)
@@ -377,7 +386,7 @@ def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optiona
                 update = Update(
                     state=DataSourceState.OFF,
                     error=error_info,
-                    revert_to_fdv1=True,
+                    fallback_to_fdv1=True,
                     environment_id=envid,
                 )
                 self.stop()
@@ -394,7 +403,7 @@ def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optiona
                     else DataSourceState.OFF
                 ),
                 error=error_info,
-                revert_to_fdv1=False,
+                fallback_to_fdv1=False,
                 environment_id=envid,
             )
 
@@ -416,7 +425,7 @@ def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optiona
             error=DataSourceErrorInfo(
                 DataSourceErrorKind.UNKNOWN, 0, time(), str(error)
             ),
-            revert_to_fdv1=False,
+            fallback_to_fdv1=False,
             environment_id=envid,
         )
         # no stacktrace here because, for a typical connection error, it'll

ldclient/impl/datasystem/fdv2.py

@@ -18,9 +18,10 @@
 from ldclient.impl.listeners import Listeners
 from ldclient.impl.repeating_task import RepeatingTask
 from ldclient.impl.rwlock import ReadWriteLock
-from ldclient.impl.util import _Fail, log
+from ldclient.impl.util import _LD_FD_FALLBACK_HEADER, _Fail, log
 from ldclient.interfaces import (
     DataSourceErrorInfo,
+    DataSourceErrorKind,
     DataSourceState,
     DataSourceStatus,
     DataSourceStatusProvider,
@@ -276,7 +277,7 @@ class ConditionDirective(str, Enum):
 
     FDV1 = "fdv1"
     """
-    FDV1 suggests that we should immediately revert to the FDv1 fallback synchronizer.
+    FDV1 suggests that we should immediately fall back to the FDv1 Fallback Synchronizer.
     """
 
 
@@ -403,7 +404,26 @@ def _run_main_loop(self, set_on_ready: Event):
             )
 
             # Run initializers first
-            self._run_initializers(set_on_ready)
+            fallback_requested = self._run_initializers(set_on_ready)
+
+            # If an initializer asked the SDK to fall back to FDv1, halt the
+            # configured FDv2 chain and switch terminally to the FDv1 Fallback
+            # Synchronizer (or transition to OFF if none is configured).
+            if fallback_requested:
+                if self._fdv1_fallback_synchronizer_builder is not None:
+                    log.warning("Falling back to FDv1 protocol")
+                    self._synchronizers = [self._fdv1_fallback_synchronizer_builder]
+                else:
+                    log.warning(
+                        "Initializer requested FDv1 fallback but none configured"
+                    )
+                    self._synchronizers = []
+                    self._data_source_status_provider.update_status(
+                        DataSourceState.OFF,
+                        self._data_source_status_provider.status.error,
+                    )
+                    set_on_ready.set()
+                    return
 
             # Run synchronizers
             self._run_synchronizers(set_on_ready)
@@ -414,14 +434,22 @@ def _run_main_loop(self, set_on_ready: Event):
             if not set_on_ready.is_set():
                 set_on_ready.set()
 
-    def _run_initializers(self, set_on_ready: Event):
-        """Run initializers to get initial data."""
+    def _run_initializers(self, set_on_ready: Event) -> bool:
+        """
+        Run initializers to get initial data.
+
+        Returns True when an initializer requested the FDv1 Fallback Directive
+        (via the X-LD-FD-Fallback response header). When that happens, any
+        accompanying payload is applied first so evaluations can serve the
+        server-provided data while the FDv1 synchronizer spins up; the caller
+        is then responsible for switching to the FDv1 Fallback Synchronizer.
+        """
         if self._data_system_config.initializers is None:
-            return
+            return False
 
         for initializer_builder in self._data_system_config.initializers:
             if self._stop_event.is_set():
-                return
+                return False
 
             try:
                 initializer = initializer_builder.build(self._config)
@@ -431,6 +459,25 @@ def _run_initializers(self, set_on_ready: Event):
 
                 if isinstance(basis_result, _Fail):
                     log.warning("Initializer %s failed: %s", initializer.name, basis_result.error)
+                    # An error response can still carry the FDv1 fallback directive.
+                    if basis_result.headers is not None and \
+                            basis_result.headers.get(_LD_FD_FALLBACK_HEADER) == 'true':
+                        log.warning(
+                            "Initializer %s requested fallback to FDv1 protocol",
+                            initializer.name,
+                        )
+                        # Surface the underlying error on the status so
+                        # programmatic monitors can see why FDv2 shut down.
+                        self._data_source_status_provider.update_status(
+                            DataSourceState.INITIALIZING,
+                            DataSourceErrorInfo(
+                                kind=DataSourceErrorKind.UNKNOWN,
+                                status_code=0,
+                                time=time.time(),
+                                message=basis_result.error,
+                            ),
+                        )
+                        return True
                     continue
 
                 basis = basis_result.value
@@ -442,9 +489,19 @@ def _run_initializers(self, set_on_ready: Event):
                 # Set ready event if an only if a selector is defined for the changeset
                 if basis.change_set.selector.is_defined():
                     set_on_ready.set()
-                    return
+
+                if basis.fallback_to_fdv1:
+                    log.warning(
+                        "Initializer %s requested fallback to FDv1 protocol",
+                        initializer.name,
+                    )
+                    return True
+
+                if basis.change_set.selector.is_defined():
+                    return False
             except Exception as e:
                 log.error("Initializer failed with exception: %s", e)
+        return False
 
     def _run_synchronizers(self, set_on_ready: Event):
         """Run synchronizers to keep data up-to-date."""
@@ -476,12 +533,12 @@ def synchronizer_loop(self: 'FDv2'):
 
                         if directive == ConditionDirective.FDV1:
                             # Abandon all synchronizers and use only fdv1 fallback
-                            log.info("Reverting to FDv1 fallback synchronizer")
+                            log.warning("Falling back to FDv1 protocol")
                             if self._fdv1_fallback_synchronizer_builder is not None:
                                 synchronizers_list = [self._fdv1_fallback_synchronizer_builder]
                                 current_index = 0
                             else:
-                                log.warning("No FDv1 fallback synchronizer available")
+                                log.warning("Synchronizer requested FDv1 fallback but none configured")
                                 synchronizers_list = []
                                 self._data_source_status_provider.update_status(
                                     DataSourceState.OFF,
@@ -608,8 +665,11 @@ def reader(self: 'FDv2'):
                 # Update status
                 self._data_source_status_provider.update_status(update.state, update.error)
 
-                # Check if we should revert to FDv1 immediately
-                if update.revert_to_fdv1:
+                # Check if we should fall back to FDv1 immediately. fallback_to_fdv1
+                # may ride along on a Valid update (payload + directive in the same
+                # response), in which case the ChangeSet has already been applied
+                # above before we hand off.
+                if update.fallback_to_fdv1:
                     return ConditionDirective.FDV1
 
                 # Check for OFF state indicating permanent failure

ldclient/interfaces.py

@@ -1366,6 +1366,13 @@ class Basis:
     change_set: ChangeSet
     persist: bool
     environment_id: Optional[str] = None
+    fallback_to_fdv1: bool = False
+    """
+    Indicates that the LaunchDarkly server has directed the SDK to fall
+    back to the FDv1 Fallback Synchronizer (via the X-LD-FD-Fallback
+    response header). When True, callers must apply ``change_set`` first
+    and then terminally switch to the FDv1 Fallback Synchronizer.
+    """
 
 
 class ChangeSetBuilder:
@@ -1647,7 +1654,14 @@ class Update:
     state: DataSourceState
     change_set: Optional[ChangeSet] = None
     error: Optional[DataSourceErrorInfo] = None
-    revert_to_fdv1: bool = False
+    fallback_to_fdv1: bool = False
+    """
+    Indicates that the LaunchDarkly server has directed the SDK to fall
+    back to the FDv1 Fallback Synchronizer (via the X-LD-FD-Fallback
+    response header). When True, callers must apply any accompanying
+    ``change_set`` first and then terminally switch to the FDv1 Fallback
+    Synchronizer. The flag may ride along on Valid or Off updates.
+    """
     environment_id: Optional[str] = None