fix: preserve FDv1 fallback signal across streaming error paths

When a streaming Start action latched the FDv1 Fallback Directive
(via X-LD-FD-Fallback: true), a subsequent Fault, JSONDecodeError, or
generic exception path could yield an Update with
fallback_to_fdv1=False because each error path independently checked
the failure's headers for the directive instead of consulting the
latched state from the earlier Start. The most visible symptom is an
HTTPStatusError 401 mid-stream after the directive was set: the SDK
treated it as an ordinary unrecoverable failure (REMOVE) instead of
honoring the directive (FDV1).

Funnel every error-path Update through a small _with_fallback_signal
helper that propagates the latched signal, and break out of the SSE
read loop unconditionally once the directive is set so we don't keep
retrying the FDv2 endpoint -- the directive is one-way and terminal.

Author: keelerm84

ldclient/impl/datasourcev2/streaming.py

@@ -213,8 +213,25 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, 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.
+        # while FDv1 takes over. The latch is one-way and terminal: once
+        # set, any subsequent payload-completing event or error must carry
+        # the signal forward and halt the stream, even if the failure path
+        # itself doesn't see the directive header.
         fallback_requested = False
+
+        def _with_fallback_signal(update: Update) -> Update:
+            """Return ``update`` decorated with ``fallback_to_fdv1=True`` when
+            the directive has been latched. Idempotent if already set."""
+            if not fallback_requested or update.fallback_to_fdv1:
+                return update
+            return Update(
+                state=update.state,
+                change_set=update.change_set,
+                error=update.error,
+                fallback_to_fdv1=True,
+                environment_id=update.environment_id,
+            )
+
         for action in self._sse.all:
             if isinstance(action, Fault):
                 # If the SSE client detects the stream has closed, then it will
@@ -228,9 +245,12 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
 
                 (update, should_continue) = self._handle_error(action.error, envid)
                 if update is not None:
-                    yield update
+                    yield _with_fallback_signal(update)
 
-                if not should_continue:
+                # The FDv1 Fallback Directive is one-way and terminal: if it
+                # was latched on a prior Start, we must not keep retrying the
+                # FDv2 endpoint even when the failure itself looks recoverable.
+                if fallback_requested or not should_continue:
                     break
                 continue
 
@@ -248,16 +268,10 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
                     self._record_stream_init(False)
                     self._connection_attempt_start_time = None
                     if fallback_requested:
-                        # Decorate the completed update with the fallback signal,
+                        # The completed update is the natural moment to honor
+                        # the latched directive: yield once with the 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
+                        yield _with_fallback_signal(update)
                         break
                     yield update
             except json.decoder.JSONDecodeError as e:
@@ -268,23 +282,25 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
 
                 (update, should_continue) = self._handle_error(e, envid)
                 if update is not None:
-                    yield update
-                if not should_continue:
+                    yield _with_fallback_signal(update)
+                if fallback_requested or not should_continue:
                     break
             except Exception as e:  # pylint: disable=broad-except
                 log.info(
                     "Error while handling stream event; will restart stream: %s", e
                 )
                 self._sse.interrupt()
 
-                yield Update(
+                yield _with_fallback_signal(Update(
                     state=DataSourceState.INTERRUPTED,
                     error=DataSourceErrorInfo(
                         DataSourceErrorKind.UNKNOWN, 0, time(), str(e)
                     ),
                     fallback_to_fdv1=False,
                     environment_id=envid,
-                )
+                ))
+                if fallback_requested:
+                    break
 
         self._sse.close()
         # Force-close the underlying urllib3 pool. SSEClient.close() only does a

ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py

@@ -626,6 +626,73 @@ def test_fallback_header_with_payload_emits_valid_with_fallback(events):  # pyli
     assert len(updates[0].change_set.changes) == 1
 
 
+def test_fallback_latched_on_start_carries_through_unrecoverable_fault():
+    """Once a Start latches the FDv1 directive, an unrecoverable Fault that
+    follows must propagate the directive even when the error itself does not
+    carry the header. The directive is one-way and terminal, so the latched
+    state from the original Start drives the Update emitted on shutdown --
+    losing it would silently strand the consumer on FDv2 instead of handing
+    off to the FDv1 Fallback Synchronizer."""
+    start_action = Start(headers={_LD_FD_FALLBACK_HEADER: 'true'})
+    # 401 is unrecoverable and the error carries no fallback header itself.
+    error = HTTPStatusError(401)
+    fault_action = Fault(error=error)
+
+    builder = list_sse_client([start_action, fault_action])
+
+    synchronizer = make_streaming_data_source()
+    synchronizer._sse_client_builder = builder
+    updates = list(synchronizer.sync(MockSelectorStore(Selector.no_selector())))
+
+    assert len(updates) == 1
+    assert updates[0].state == DataSourceState.OFF
+    assert updates[0].fallback_to_fdv1 is True
+    assert updates[0].error is not None
+    assert updates[0].error.status_code == 401
+
+
+def test_fallback_latched_on_start_carries_through_recoverable_fault():
+    """A recoverable Fault arriving after the directive was latched must also
+    propagate the signal and halt the stream -- the directive overrides the
+    ordinary retry policy because it is terminal."""
+    start_action = Start(headers={_LD_FD_FALLBACK_HEADER: 'true'})
+    # 408 is recoverable; without the latch we would retry transparently.
+    fault_action = Fault(error=HTTPStatusError(408))
+
+    builder = list_sse_client([start_action, fault_action])
+
+    synchronizer = make_streaming_data_source()
+    synchronizer._sse_client_builder = builder
+    updates = list(synchronizer.sync(MockSelectorStore(Selector.no_selector())))
+
+    assert len(updates) == 1
+    assert updates[0].fallback_to_fdv1 is True
+    # 408 is recoverable so the Update from _handle_error is INTERRUPTED, but
+    # the latched directive must still drive the consumer to FDv1.
+    assert updates[0].state == DataSourceState.INTERRUPTED
+
+
+def test_fallback_latched_on_start_carries_through_malformed_event(events):  # pylint: disable=redefined-outer-name
+    """A malformed event (JSONDecodeError) after the directive was latched
+    must propagate the signal on the resulting Interrupted Update."""
+    bad_event = Event(event=EventName.PUT_OBJECT, data="not valid json")
+    start_action = Start(headers={_LD_FD_FALLBACK_HEADER: 'true'})
+
+    builder = list_sse_client([start_action, events[EventName.SERVER_INTENT], bad_event])
+
+    synchronizer = make_streaming_data_source()
+    synchronizer._sse_client_builder = builder
+    updates = list(synchronizer.sync(MockSelectorStore(Selector.no_selector())))
+
+    # The malformed-event update must surface the latched directive so the
+    # consumer can hand off to FDv1 instead of trying to keep the FDv2 stream.
+    assert len(updates) == 1
+    assert updates[0].state == DataSourceState.INTERRUPTED
+    assert updates[0].fallback_to_fdv1 is True
+    assert updates[0].error is not None
+    assert updates[0].error.kind == DataSourceErrorKind.INVALID_DATA
+
+
 def test_streaming_closes_underlying_pool_on_fallback(events):  # pylint: disable=redefined-outer-name
     """When the FDv1 Fallback Directive engages, the underlying urllib3
     connection pool must be torn down so the FDv2 streaming TCP connection