From d7e85617705c03da8e858576489755917962222a Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 19 Oct 2026 17:27:47 +0000 Subject: [PATCH] fix: Make FDv1 fallback TTL-based and always recover to FDv2 Co-Authored-By: Bee Klimt --- ldclient/impl/datasourcev2/async_polling.py | 4 +- ldclient/impl/datasourcev2/async_streaming.py | 42 ++-- ldclient/impl/datasourcev2/polling.py | 4 +- ldclient/impl/datasourcev2/polling_common.py | 61 +++++- ldclient/impl/datasourcev2/streaming.py | 47 ++-- .../impl/datasourcev2/streaming_common.py | 28 ++- ldclient/impl/datasystem/async_fdv2.py | 201 ++++++++++++------ ldclient/impl/datasystem/fdv2.py | 200 +++++++++++------ ldclient/impl/datasystem/protocolv2.py | 13 +- ldclient/impl/util.py | 30 +++ ldclient/interfaces.py | 20 +- .../impl/datasourcev2/test_async_streaming.py | 41 +++- .../test_polling_payload_parsing.py | 41 +++- .../datasourcev2/test_polling_synchronizer.py | 33 +++ .../test_streaming_synchronizer.py | 66 +++++- .../impl/datasystem/test_async_fdv2.py | 81 ++++++- .../impl/datasystem/test_fdv2_datasystem.py | 120 +++++++---- ldclient/testing/impl/test_util.py | 44 +++- 18 files changed, 814 insertions(+), 262 deletions(-) diff --git a/ldclient/impl/datasourcev2/async_polling.py b/ldclient/impl/datasourcev2/async_polling.py index e3b6a3a3..49a14569 100644 --- a/ldclient/impl/datasourcev2/async_polling.py +++ b/ldclient/impl/datasourcev2/async_polling.py @@ -24,7 +24,8 @@ fdv1_polling_payload_to_changeset, map_polling_result, polling_payload_to_changeset, - polling_result_to_basis + polling_result_to_basis, + with_goodbye_fallback ) from ldclient.impl.http import ASYNC_USER_AGENT from ldclient.impl.util import ( @@ -228,6 +229,7 @@ async def fetch(self, selector: Optional[Selector]) -> PollingResult: etag, ) + headers = with_goodbye_fallback(headers, data) changeset_result = polling_payload_to_changeset(data) if isinstance(changeset_result, _Success): return _Success(value=(changeset_result.value, headers)) diff --git a/ldclient/impl/datasourcev2/async_streaming.py b/ldclient/impl/datasourcev2/async_streaming.py index 36e7d09a..0498d2dc 100644 --- a/ldclient/impl/datasourcev2/async_streaming.py +++ b/ldclient/impl/datasourcev2/async_streaming.py @@ -24,7 +24,11 @@ with_fallback_signal ) from ldclient.impl.datasystem import DiagnosticAccumulator, DiagnosticSource -from ldclient.impl.util import _LD_ENVID_HEADER, _LD_FD_FALLBACK_HEADER, log +from ldclient.impl.util import ( + _LD_ENVID_HEADER, + fdv1_fallback_ttl_from_headers, + log +) from ldclient.interfaces import ( AsyncSynchronizer, ChangeSetBuilder, @@ -153,12 +157,12 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: self._connection_attempt_start_time = time() envid = None - # A Start action with the X-LD-FD-Fallback: true header sets fallback_requested. + # A Start action with the X-LD-FD-Fallback: true header sets fallback_ttl. # We apply the current payload before halting, so consumers can serve that data - # while FDv1 takes over. Once set, the flag stays set: every later payload event + # while FDv1 takes over. Once set, it stays set: every later payload event # or error re-emits the signal and halts the stream, even when that event has no # directive header of its own. - fallback_requested = False + fallback_ttl: Optional[float] = None try: async for action in self._sse.all: @@ -174,19 +178,18 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: (update, should_continue) = await self._handle_error(action.error, envid) if update is not None: - yield with_fallback_signal(update, fallback_requested) + yield with_fallback_signal(update, fallback_ttl) - # The FDv1 Fallback Directive is one-way and terminal: if it - # was latched on a prior Start, we must not keep retrying the + # Once the FDv1 Fallback Directive 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: + if fallback_ttl is not None or not should_continue: break continue if isinstance(action, Start) and action.headers is not None: envid = action.headers.get(_LD_ENVID_HEADER, envid) - if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true': - fallback_requested = True + fallback_ttl = fdv1_fallback_ttl_from_headers(action.headers) or fallback_ttl if not isinstance(action, Event): continue @@ -196,13 +199,12 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: if update is not None: self._record_stream_init(False) self._connection_attempt_start_time = None - if fallback_requested: - # 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. - yield with_fallback_signal(update, fallback_requested) - break + update = with_fallback_signal(update, fallback_ttl) yield update + # Honor a latched or in-band directive once the update is out: + # halt, and the consumer will switch to FDv1. + if update.fallback_to_fdv1: + break except json.decoder.JSONDecodeError as e: log.info( "Error while handling stream event; will restart stream: %s", e @@ -211,8 +213,8 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: (update, should_continue) = await self._handle_error(e, envid) if update is not None: - yield with_fallback_signal(update, fallback_requested) - if fallback_requested or not should_continue: + yield with_fallback_signal(update, fallback_ttl) + if fallback_ttl is not None or not should_continue: break except Exception as e: # pylint: disable=broad-except log.info( @@ -227,8 +229,8 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: ), fallback_to_fdv1=False, environment_id=envid, - ), fallback_requested) - if fallback_requested: + ), fallback_ttl) + if fallback_ttl is not None: break finally: await self._sse.close() diff --git a/ldclient/impl/datasourcev2/polling.py b/ldclient/impl/datasourcev2/polling.py index a96d5ed0..c4a3c6f6 100644 --- a/ldclient/impl/datasourcev2/polling.py +++ b/ldclient/impl/datasourcev2/polling.py @@ -23,7 +23,8 @@ fdv1_polling_payload_to_changeset, map_polling_result, polling_payload_to_changeset, - polling_result_to_basis + polling_result_to_basis, + with_goodbye_fallback ) from ldclient.impl.http import HTTPFactory, _base_headers from ldclient.impl.util import ( @@ -212,6 +213,7 @@ def fetch(self, selector: Optional[Selector]) -> PollingResult: etag, ) + headers = with_goodbye_fallback(headers, data) changeset_result = polling_payload_to_changeset(data) if isinstance(changeset_result, _Success): return _Success(value=(changeset_result.value, headers)) diff --git a/ldclient/impl/datasourcev2/polling_common.py b/ldclient/impl/datasourcev2/polling_common.py index 532efd9b..b6b85d6c 100644 --- a/ldclient/impl/datasourcev2/polling_common.py +++ b/ldclient/impl/datasourcev2/polling_common.py @@ -6,20 +6,25 @@ from dataclasses import dataclass from enum import Enum from time import time -from typing import Mapping, Tuple +from typing import Mapping, Optional, Tuple + +from urllib3._collections import HTTPHeaderDict from ldclient.impl.datasystem.protocolv2 import ( DeleteObject, EventName, + Goodbye, PutObject ) from ldclient.impl.util import ( _LD_ENVID_HEADER, _LD_FD_FALLBACK_HEADER, + _LD_FD_FALLBACK_TTL_HEADER, UnsuccessfulResponseException, _Fail, _Result, _Success, + fdv1_fallback_ttl_from_headers, http_error_message, is_http_error_recoverable, log @@ -40,6 +45,31 @@ ) +def with_goodbye_fallback(headers: Mapping, data: dict) -> Mapping: + """ + Return ``headers`` with the FDv1 fallback directive added when the polling + payload contains a goodbye event carrying protocolFallbackTTL, so the + in-band directive is handled like the X-LD-FD-Fallback header. + """ + events = data.get("events") + if not isinstance(events, list): + return headers + + for event in events: + if not isinstance(event, dict) or event.get("event") != EventName.GOODBYE: + continue + ttl = event.get("data", {}).get("protocolFallbackTTL") if isinstance(event.get("data"), dict) else None + if ttl is None: + continue + + merged = HTTPHeaderDict(headers) + merged[_LD_FD_FALLBACK_HEADER] = 'true' + merged[_LD_FD_FALLBACK_TTL_HEADER] = str(ttl) if isinstance(ttl, int) and not isinstance(ttl, bool) else '' + return merged + + return headers + + # pylint: disable=too-many-branches,too-many-return-statements def polling_payload_to_changeset(data: dict) -> _Result[ChangeSet, str]: """ @@ -49,6 +79,7 @@ def polling_payload_to_changeset(data: dict) -> _Result[ChangeSet, str]: return _Fail(error="Invalid payload: 'events' key is missing or not a list") builder = ChangeSetBuilder() + goodbye_received = False for event in data["events"]: if not isinstance(event, dict): @@ -83,6 +114,14 @@ def polling_payload_to_changeset(data: dict) -> _Result[ChangeSet, str]: builder.add_delete( delete_object.kind, delete_object.key, delete_object.version ) + elif event["event"] == EventName.GOODBYE: + try: + goodbye = Goodbye.from_dict(event["data"]) + except ValueError as err: + return _Fail(error="Invalid JSON in goodbye", exception=err) + + log.info("Polling response contained goodbye: %s", goodbye.reason) + goodbye_received = True elif event["event"] == EventName.PAYLOAD_TRANSFERRED: try: selector = Selector.from_dict(event["data"]) @@ -94,6 +133,9 @@ def polling_payload_to_changeset(data: dict) -> _Result[ChangeSet, str]: error="Invalid JSON in payload transferred object", exception=err ) + if goodbye_received: + return _Success(ChangeSetBuilder.no_changes()) + return _Fail(error="didn't receive any known protocol events in polling payload") @@ -170,11 +212,13 @@ def polling_result_to_basis( if not isinstance(env_id, str): env_id = None + fallback_ttl = fdv1_fallback_ttl_from_headers(headers) basis = Basis( change_set=change_set, persist=change_set.selector.is_defined(), environment_id=env_id, - fallback_to_fdv1=headers.get(_LD_FD_FALLBACK_HEADER) == 'true', + fallback_to_fdv1=fallback_ttl is not None, + fallback_ttl=fallback_ttl, ) return _Success(value=basis) @@ -212,11 +256,10 @@ def map_polling_result( polling synchronizers. """ if isinstance(result, _Fail): - fallback = None + fallback_ttl = fdv1_fallback_ttl_from_headers(result.headers) envid = None if result.headers is not None: - fallback = result.headers.get(_LD_FD_FALLBACK_HEADER) == 'true' envid = result.headers.get(_LD_ENVID_HEADER) if isinstance(result.exception, UnsuccessfulResponseException): @@ -229,13 +272,14 @@ def map_polling_result( ), ) - if fallback: + if fallback_ttl is not None: return PollDecision( Update( state=DataSourceState.OFF, error=error_info, fallback_to_fdv1=True, environment_id=envid, + fallback_ttl=fallback_ttl, ), PollAction.BREAK, ) @@ -269,13 +313,14 @@ def map_polling_result( # 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: + if fallback_ttl is not None: return PollDecision( Update( state=DataSourceState.OFF, error=error_info, fallback_to_fdv1=True, environment_id=envid, + fallback_ttl=fallback_ttl, ), PollAction.BREAK, ) @@ -290,12 +335,14 @@ def map_polling_result( ) (change_set, headers) = result.value + fallback_ttl = fdv1_fallback_ttl_from_headers(headers) return PollDecision( Update( state=DataSourceState.VALID, change_set=change_set, environment_id=headers.get(_LD_ENVID_HEADER), - fallback_to_fdv1=headers.get(_LD_FD_FALLBACK_HEADER) == 'true', + fallback_to_fdv1=fallback_ttl is not None, + fallback_ttl=fallback_ttl, ), PollAction.WAIT_BREAK_IF_SET, ) diff --git a/ldclient/impl/datasourcev2/streaming.py b/ldclient/impl/datasourcev2/streaming.py index 16e533c1..67eeef89 100644 --- a/ldclient/impl/datasourcev2/streaming.py +++ b/ldclient/impl/datasourcev2/streaming.py @@ -28,7 +28,11 @@ ) from ldclient.impl.datasystem import DiagnosticAccumulator, DiagnosticSource from ldclient.impl.http import HTTPFactory, _base_headers -from ldclient.impl.util import _LD_ENVID_HEADER, _LD_FD_FALLBACK_HEADER, log +from ldclient.impl.util import ( + _LD_ENVID_HEADER, + fdv1_fallback_ttl_from_headers, + log +) from ldclient.interfaces import ( ChangeSetBuilder, DataSourceErrorInfo, @@ -161,14 +165,13 @@ 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 + # fallback_ttl 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. 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 + # while FDv1 takes over. 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_ttl: Optional[float] = None for action in self._sse.all: if isinstance(action, Fault): @@ -183,19 +186,18 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: (update, should_continue) = self._handle_error(action.error, envid) if update is not None: - yield with_fallback_signal(update, fallback_requested) + yield with_fallback_signal(update, fallback_ttl) - # The FDv1 Fallback Directive is one-way and terminal: if it - # was latched on a prior Start, we must not keep retrying the + # Once the FDv1 Fallback Directive 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: + if fallback_ttl is not None or not should_continue: break continue if isinstance(action, Start) and action.headers is not None: envid = action.headers.get(_LD_ENVID_HEADER, envid) - if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true': - fallback_requested = True + fallback_ttl = fdv1_fallback_ttl_from_headers(action.headers) or fallback_ttl if not isinstance(action, Event): continue @@ -205,13 +207,12 @@ 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: - # 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. - yield with_fallback_signal(update, fallback_requested) - break + update = with_fallback_signal(update, fallback_ttl) yield update + # Honor a latched or in-band directive once the update is out: + # halt, and the consumer will switch to FDv1. + if update.fallback_to_fdv1: + break except json.decoder.JSONDecodeError as e: log.info( "Error while handling stream event; will restart stream: %s", e @@ -220,8 +221,8 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: (update, should_continue) = self._handle_error(e, envid) if update is not None: - yield with_fallback_signal(update, fallback_requested) - if fallback_requested or not should_continue: + yield with_fallback_signal(update, fallback_ttl) + if fallback_ttl is not None or not should_continue: break except Exception as e: # pylint: disable=broad-except log.info( @@ -236,8 +237,8 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: ), fallback_to_fdv1=False, environment_id=envid, - ), fallback_requested) - if fallback_requested: + ), fallback_ttl) + if fallback_ttl is not None: break self._sse.close() diff --git a/ldclient/impl/datasourcev2/streaming_common.py b/ldclient/impl/datasourcev2/streaming_common.py index cb0428d8..fa918981 100644 --- a/ldclient/impl/datasourcev2/streaming_common.py +++ b/ldclient/impl/datasourcev2/streaming_common.py @@ -20,7 +20,8 @@ ) from ldclient.impl.util import ( _LD_ENVID_HEADER, - _LD_FD_FALLBACK_HEADER, + fdv1_fallback_ttl, + fdv1_fallback_ttl_from_headers, http_error_message, is_http_error_recoverable, log @@ -78,7 +79,15 @@ def process_message( goodbye = Goodbye.from_dict(json.loads(msg.data)) log.info("SSE server sent goodbye: %s", goodbye.reason) - return None + if goodbye.protocol_fallback_ttl is None: + return None + + return Update( + state=DataSourceState.INTERRUPTED, + fallback_to_fdv1=True, + fallback_ttl=fdv1_fallback_ttl(goodbye.protocol_fallback_ttl), + environment_id=envid, + ) if msg.event == EventName.ERROR: error = Error.from_dict(json.loads(msg.data)) @@ -108,13 +117,13 @@ def process_message( return None -def with_fallback_signal(update: Update, fallback_requested: bool) -> Update: +def with_fallback_signal(update: Update, fallback_ttl: Optional[float]) -> Update: """ - Return ``update`` marked with ``fallback_to_fdv1=True`` when the FDv1 - fallback directive has been latched. Returns the update unchanged when the - directive is not set or the update already carries it. + Return ``update`` marked with the FDv1 fallback directive when one has + been latched with the given TTL. Returns the update unchanged when no + directive is latched or the update already carries its own. """ - if not fallback_requested or update.fallback_to_fdv1: + if fallback_ttl is None or update.fallback_to_fdv1: return update return Update( state=update.state, @@ -122,6 +131,7 @@ def with_fallback_signal(update: Update, fallback_requested: bool) -> Update: error=update.error, fallback_to_fdv1=True, environment_id=update.environment_id, + fallback_ttl=fallback_ttl, ) @@ -183,12 +193,14 @@ def classify_stream_error( if envid is None and error.headers is not None: envid = error.headers.get(_LD_ENVID_HEADER) - if error.headers is not None and error.headers.get(_LD_FD_FALLBACK_HEADER) == 'true': + fallback_ttl = fdv1_fallback_ttl_from_headers(error.headers) + if fallback_ttl is not None: update = Update( state=DataSourceState.OFF, error=error_info, fallback_to_fdv1=True, environment_id=envid, + fallback_ttl=fallback_ttl, ) return StreamErrorDecision( update=update, diff --git a/ldclient/impl/datasystem/async_fdv2.py b/ldclient/impl/datasystem/async_fdv2.py index d6b6c3e3..fde868a6 100644 --- a/ldclient/impl/datasystem/async_fdv2.py +++ b/ldclient/impl/datasystem/async_fdv2.py @@ -7,7 +7,7 @@ import asyncio import inspect import time -from typing import Any, Callable, Dict, List, Mapping, Optional +from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple from ldclient.async_config import AsyncConfig, AsyncDataSystemConfig from ldclient.config import DataSourceBuilder @@ -39,7 +39,12 @@ ) from ldclient.impl.datasystem.store import _decode from ldclient.impl.listeners import Listeners -from ldclient.impl.util import _LD_FD_FALLBACK_HEADER, _Fail, log +from ldclient.impl.util import ( + _Fail, + fdv1_fallback_ttl, + fdv1_fallback_ttl_from_headers, + log +) from ldclient.interfaces import ( AsyncFeatureStore, AsyncReadOnlyStore, @@ -342,30 +347,12 @@ async def _run_main_loop(self, set_on_ready: AsyncEvent): DataSourceState.INITIALIZING, None ) - # Run initializers first - fallback_requested = await 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 initializers first. If one directs the SDK to fall back to + # FDv1, the synchronizer loop starts in FDv1 fallback. + fallback_ttl = await self._run_initializers(set_on_ready) # Run synchronizers - await self._run_synchronizers(set_on_ready) + await self._run_synchronizers(set_on_ready, fallback_ttl) except Exception as e: log.error("Error in AsyncFDv2 main loop: %s", e) @@ -373,22 +360,22 @@ async def _run_main_loop(self, set_on_ready: AsyncEvent): if not set_on_ready.is_set(): set_on_ready.set() - async def _run_initializers(self, set_on_ready: AsyncEvent) -> bool: + async def _run_initializers(self, set_on_ready: AsyncEvent) -> Optional[float]: """ 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. + Returns the FDv1 fallback TTL, in seconds, when an initializer + requested the FDv1 Fallback Directive, or None otherwise. 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 falling back to FDv1 for the TTL. """ if self._data_system_config.initializers is None: - return False + return None for initializer_builder in self._data_system_config.initializers: if self._stop_event.is_set(): - return False + return None try: initializer = initializer_builder.build(self._config) @@ -399,8 +386,8 @@ async def _run_initializers(self, set_on_ready: AsyncEvent) -> bool: 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': + fallback_ttl = fdv1_fallback_ttl_from_headers(basis_result.headers) + if fallback_ttl is not None: log.warning( "Initializer %s requested fallback to FDv1 protocol", initializer.name, @@ -416,7 +403,7 @@ async def _run_initializers(self, set_on_ready: AsyncEvent) -> bool: message=basis_result.error, ), ) - return True + return fallback_ttl continue basis = basis_result.value @@ -435,35 +422,54 @@ async def _run_initializers(self, set_on_ready: AsyncEvent) -> bool: "Initializer %s requested fallback to FDv1 protocol", initializer.name, ) - return True + return basis.fallback_ttl if basis.fallback_ttl is not None else fdv1_fallback_ttl() if selector_defined: - return False + return None except Exception as e: log.error("Initializer failed with exception: %s", e) - return False + return None + + async def _run_synchronizers(self, set_on_ready: AsyncEvent, fallback_ttl: Optional[float] = None): + """ + Run synchronizers to keep data up-to-date. - async def _run_synchronizers(self, set_on_ready: AsyncEvent): - """Run synchronizers to keep data up-to-date.""" + :param fallback_ttl: when set, start in FDv1 fallback for this many seconds + """ # If no synchronizers configured, just set ready and return - if len(self._synchronizers) == 0: + if len(self._synchronizers) == 0 and fallback_ttl is None: set_on_ready.set() return self._runner.spawn( "AsyncFDv2-synchronizers", - lambda: self._synchronizer_loop(set_on_ready), + lambda: self._synchronizer_loop(set_on_ready, fallback_ttl), ) - async def _synchronizer_loop(self, set_on_ready: AsyncEvent): + async def _synchronizer_loop(self, set_on_ready: AsyncEvent, fallback_ttl: Optional[float] = None): try: - # Make a working copy of the synchronizers list + # Make a working copy of the synchronizers list. self._synchronizers + # stays untouched so FDv2 can be restored after an FDv1 fallback. synchronizers_list = list(self._synchronizers) current_index = 0 # Always ensure ready event is set when we exit - while not self._stop_event.is_set() and len(synchronizers_list) > 0: + while not self._stop_event.is_set(): try: + if fallback_ttl is not None: + await self._run_fdv1_fallback(set_on_ready, fallback_ttl) + fallback_ttl = None + + if self._stop_event.is_set(): + break + log.info("FDv1 fallback TTL expired; returning to FDv2") + synchronizers_list = list(self._synchronizers) + current_index = 0 + continue + + if len(synchronizers_list) == 0: + break + async with self._lock: synchronizer: AsyncSynchronizer = synchronizers_list[current_index].build(self._config) self._active_synchronizer = synchronizer @@ -472,24 +478,13 @@ async def _synchronizer_loop(self, set_on_ready: AsyncEvent): log.info("Synchronizer %s (index %d) is starting", synchronizer.name, current_index) - directive = await self._consume_synchronizer_results( + (directive, fallback_ttl) = await self._consume_synchronizer_results( synchronizer, set_on_ready, current_index != 0 ) if directive == ConditionDirective.FDV1: - # Abandon all synchronizers and use only fdv1 fallback - 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("Synchronizer requested FDv1 fallback but none configured") - synchronizers_list = [] - self._data_source_status_provider.update_status( - DataSourceState.OFF, - self._data_source_status_provider.status.error - ) - break + # The synchronizer has already been stopped; fall + # back to FDv1 on the next iteration. continue elif directive == ConditionDirective.REMOVE: # Permanent failure - remove synchronizer from list @@ -534,16 +529,71 @@ async def _synchronizer_loop(self, set_on_ready: AsyncEvent): await self._active_synchronizer.stop() self._active_synchronizer = None + async def _run_fdv1_fallback(self, set_on_ready: AsyncEvent, ttl: float): + """ + Serve data from the FDv1 Fallback Synchronizer, if one is configured, + for ``ttl`` seconds or until the data system is stopped. Without one, + just wait out the TTL. The caller then returns to FDv2. + + Synchronizers run one at a time on the calling task, and each is + stopped before the next starts, so FDv1 and FDv2 never write to the + store concurrently. + """ + log.warning("Falling back to FDv1 protocol; will retry FDv2 in %.0f seconds", ttl) + deadline = time.monotonic() + ttl + + if self._fdv1_fallback_synchronizer_builder is None: + log.warning("FDv1 fallback requested but no FDv1 fallback synchronizer is configured") + self._data_source_status_provider.update_status( + DataSourceState.INTERRUPTED, + self._data_source_status_provider.status.error or DataSourceErrorInfo( + kind=DataSourceErrorKind.UNKNOWN, + status_code=0, + time=time.time(), + message="FDv1 fallback directive received; no FDv1 fallback configured", + ), + ) + set_on_ready.set() + await self._stop_event.wait(ttl) + return + + while not self._stop_event.is_set(): + remaining = deadline - time.monotonic() + if remaining <= 0: + return + + async with self._lock: + synchronizer: AsyncSynchronizer = self._fdv1_fallback_synchronizer_builder.build(self._config) + self._active_synchronizer = synchronizer + if isinstance(synchronizer, DiagnosticSource) and self._diagnostic_accumulator is not None: + synchronizer.set_diagnostic_accumulator(self._diagnostic_accumulator) + + log.info("FDv1 fallback synchronizer %s is starting", synchronizer.name) + + (directive, _) = await self._consume_synchronizer_results( + synchronizer, set_on_ready, False, remaining + ) + + if directive == ConditionDirective.RECOVER: + return + if directive == ConditionDirective.REMOVE: + # FDv1 failed permanently; FDv2 is still off until the TTL expires. + await self._stop_event.wait(max(0.0, deadline - time.monotonic())) + return + async def _consume_synchronizer_results( self, synchronizer: AsyncSynchronizer, set_on_ready: AsyncEvent, check_recovery: bool, - ) -> ConditionDirective: + fdv2_recovery_ttl: Optional[float] = None, + ) -> Tuple[ConditionDirective, Optional[float]]: """ Consume results from a synchronizer until a condition is met or it fails. - :return: the ConditionDirective describing how to proceed + :param fdv2_recovery_ttl: when set, return RECOVER after this many seconds + :return: the ConditionDirective describing how to proceed, plus the + FDv1 fallback TTL in seconds when the directive is FDV1 """ action_queue: AsyncQueue = AsyncQueue() state_age = _StateAgeTracker() @@ -561,35 +611,45 @@ async def reader(): finally: await action_queue.put("quit") + async def recovery_timer(): + await asyncio.sleep(fdv2_recovery_ttl) + await action_queue.put("recover") + sync_reader: Optional[TaskHandle] = None + recovery_task: Optional[TaskHandle] = None try: timer.start() + if fdv2_recovery_ttl is not None: + recovery_task = spawn_handle("AsyncFDv2-fdv2-recovery-timer", recovery_timer) sync_reader = spawn_handle("AsyncFDv2-sync-reader", reader) while True: # Honor a stop request every iteration so a queue that always has # an item ready cannot starve the check. if self._stop_event.is_set(): - return ConditionDirective.FALLBACK + return (ConditionDirective.FALLBACK, None) update = await action_queue.get() if isinstance(update, str): if update == "quit": break + if update == "recover": + return (ConditionDirective.RECOVER, None) + if update == "check": # Check condition periodically current_status = self._data_source_status_provider.status seconds_in_state = state_age.seconds_in_state(current_status) if check_recovery and recovery_condition(current_status, seconds_in_state): - return ConditionDirective.RECOVER + return (ConditionDirective.RECOVER, None) if fallback_condition(current_status, seconds_in_state): - return ConditionDirective.FALLBACK + return (ConditionDirective.FALLBACK, None) continue log.info("Synchronizer %s update: %s", synchronizer.name, update.state) if self._stop_event.is_set(): - return ConditionDirective.FALLBACK + return (ConditionDirective.FALLBACK, None) # Handle the update if update.change_set is not None: @@ -607,16 +667,19 @@ async def reader(): # response), in which case the ChangeSet has already been applied # above before we hand off. if update.fallback_to_fdv1: - return ConditionDirective.FDV1 + fallback_ttl = update.fallback_ttl if update.fallback_ttl is not None else fdv1_fallback_ttl() + return (ConditionDirective.FDV1, fallback_ttl) # Check for OFF state indicating permanent failure if update.state == DataSourceState.OFF: - return ConditionDirective.REMOVE + return (ConditionDirective.REMOVE, None) except Exception as e: log.error("Error consuming synchronizer results: %s", e) - return ConditionDirective.REMOVE + return (ConditionDirective.REMOVE, None) finally: timer.stop() + if recovery_task is not None: + recovery_task.cancel() if sync_reader is not None: sync_reader.cancel() @@ -627,7 +690,7 @@ async def reader(): # If we reach here, the synchronizer's iterator completed normally (no more updates) # For continuous synchronizers (streaming/polling), this is unexpected and indicates # the synchronizer can't provide more updates, so we should remove it and fall back - return ConditionDirective.REMOVE + return (ConditionDirective.REMOVE, None) @property def store(self) -> AsyncReadOnlyStore: diff --git a/ldclient/impl/datasystem/fdv2.py b/ldclient/impl/datasystem/fdv2.py index 1fbcbd33..201266af 100644 --- a/ldclient/impl/datasystem/fdv2.py +++ b/ldclient/impl/datasystem/fdv2.py @@ -1,7 +1,7 @@ import time from queue import Queue -from threading import Event, Thread -from typing import Any, Callable, Dict, List, Mapping, Optional +from threading import Event, Thread, Timer +from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple from ldclient.config import Config, DataSourceBuilder, DataSystemConfig from ldclient.feature_store import _FeatureStoreDataSetSorter @@ -19,7 +19,12 @@ from ldclient.impl.listeners import Listeners from ldclient.impl.repeating_task import RepeatingTask from ldclient.impl.rwlock import ReadWriteLock -from ldclient.impl.util import _LD_FD_FALLBACK_HEADER, _Fail, log +from ldclient.impl.util import ( + _Fail, + fdv1_fallback_ttl, + fdv1_fallback_ttl_from_headers, + log +) from ldclient.interfaces import ( DataSourceErrorInfo, DataSourceErrorKind, @@ -327,30 +332,12 @@ def _run_main_loop(self, set_on_ready: Event): DataSourceState.INITIALIZING, None ) - # Run initializers first - 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 initializers first. If one directs the SDK to fall back to + # FDv1, the synchronizer loop starts in FDv1 fallback. + fallback_ttl = self._run_initializers(set_on_ready) # Run synchronizers - self._run_synchronizers(set_on_ready) + self._run_synchronizers(set_on_ready, fallback_ttl) except Exception as e: log.error("Error in FDv2 main loop: %s", e) @@ -358,22 +345,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) -> bool: + def _run_initializers(self, set_on_ready: Event) -> Optional[float]: """ 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. + Returns the FDv1 fallback TTL, in seconds, when an initializer + requested the FDv1 Fallback Directive, or None otherwise. 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 falling back to FDv1 for the TTL. """ if self._data_system_config.initializers is None: - return False + return None for initializer_builder in self._data_system_config.initializers: if self._stop_event.is_set(): - return False + return None try: initializer = initializer_builder.build(self._config) @@ -384,8 +371,8 @@ def _run_initializers(self, set_on_ready: Event) -> bool: 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': + fallback_ttl = fdv1_fallback_ttl_from_headers(basis_result.headers) + if fallback_ttl is not None: log.warning( "Initializer %s requested fallback to FDv1 protocol", initializer.name, @@ -401,7 +388,7 @@ def _run_initializers(self, set_on_ready: Event) -> bool: message=basis_result.error, ), ) - return True + return fallback_ttl continue basis = basis_result.value @@ -422,30 +409,49 @@ def _run_initializers(self, set_on_ready: Event) -> bool: "Initializer %s requested fallback to FDv1 protocol", initializer.name, ) - return True + return basis.fallback_ttl if basis.fallback_ttl is not None else fdv1_fallback_ttl() if selector_defined: - return False + return None except Exception as e: log.error("Initializer failed with exception: %s", e) - return False + return None - def _run_synchronizers(self, set_on_ready: Event): - """Run synchronizers to keep data up-to-date.""" + def _run_synchronizers(self, set_on_ready: Event, fallback_ttl: Optional[float] = None): + """ + Run synchronizers to keep data up-to-date. + + :param fallback_ttl: when set, start in FDv1 fallback for this many seconds + """ # If no synchronizers configured, just set ready and return - if len(self._synchronizers) == 0: + if len(self._synchronizers) == 0 and fallback_ttl is None: set_on_ready.set() return - def synchronizer_loop(self: 'FDv2'): + def synchronizer_loop(self: 'FDv2', fallback_ttl: Optional[float]): try: - # Make a working copy of the synchronizers list + # Make a working copy of the synchronizers list. self._synchronizers + # stays untouched so FDv2 can be restored after an FDv1 fallback. synchronizers_list = list(self._synchronizers) current_index = 0 # Always ensure ready event is set when we exit - while not self._stop_event.is_set() and len(synchronizers_list) > 0: + while not self._stop_event.is_set(): try: + if fallback_ttl is not None: + self._run_fdv1_fallback(set_on_ready, fallback_ttl) + fallback_ttl = None + + if self._stop_event.is_set(): + break + log.info("FDv1 fallback TTL expired; returning to FDv2") + synchronizers_list = list(self._synchronizers) + current_index = 0 + continue + + if len(synchronizers_list) == 0: + break + with self._lock.write(): synchronizer: Synchronizer = synchronizers_list[current_index].build(self._config) self._active_synchronizer = synchronizer @@ -454,24 +460,13 @@ def synchronizer_loop(self: 'FDv2'): log.info("Synchronizer %s (index %d) is starting", synchronizer.name, current_index) - directive = self._consume_synchronizer_results( + (directive, fallback_ttl) = self._consume_synchronizer_results( synchronizer, set_on_ready, current_index != 0 ) if directive == ConditionDirective.FDV1: - # Abandon all synchronizers and use only fdv1 fallback - 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("Synchronizer requested FDv1 fallback but none configured") - synchronizers_list = [] - self._data_source_status_provider.update_status( - DataSourceState.OFF, - self._data_source_status_provider.status.error - ) - break + # The synchronizer has already been stopped; fall + # back to FDv1 on the next iteration. continue elif directive == ConditionDirective.REMOVE: # Permanent failure - remove synchronizer from list @@ -519,22 +514,77 @@ def synchronizer_loop(self: 'FDv2'): sync_thread = Thread( target=synchronizer_loop, name="FDv2-synchronizers", - args=(self,), + args=(self, fallback_ttl), daemon=True ) sync_thread.start() self._threads.append(sync_thread) + def _run_fdv1_fallback(self, set_on_ready: Event, ttl: float): + """ + Serve data from the FDv1 Fallback Synchronizer, if one is configured, + for ``ttl`` seconds or until the data system is stopped. Without one, + just wait out the TTL. The caller then returns to FDv2. + + Synchronizers run one at a time on the calling thread, and each is + stopped before the next starts, so FDv1 and FDv2 never write to the + store concurrently. + """ + log.warning("Falling back to FDv1 protocol; will retry FDv2 in %.0f seconds", ttl) + deadline = time.monotonic() + ttl + + if self._fdv1_fallback_synchronizer_builder is None: + log.warning("FDv1 fallback requested but no FDv1 fallback synchronizer is configured") + self._data_source_status_provider.update_status( + DataSourceState.INTERRUPTED, + self._data_source_status_provider.status.error or DataSourceErrorInfo( + kind=DataSourceErrorKind.UNKNOWN, + status_code=0, + time=time.time(), + message="FDv1 fallback directive received; no FDv1 fallback configured", + ), + ) + set_on_ready.set() + self._stop_event.wait(ttl) + return + + while not self._stop_event.is_set(): + remaining = deadline - time.monotonic() + if remaining <= 0: + return + + with self._lock.write(): + synchronizer: Synchronizer = self._fdv1_fallback_synchronizer_builder.build(self._config) + self._active_synchronizer = synchronizer + if isinstance(synchronizer, DiagnosticSource) and self._diagnostic_accumulator is not None: + synchronizer.set_diagnostic_accumulator(self._diagnostic_accumulator) + + log.info("FDv1 fallback synchronizer %s is starting", synchronizer.name) + + (directive, _) = self._consume_synchronizer_results( + synchronizer, set_on_ready, False, remaining + ) + + if directive == ConditionDirective.RECOVER: + return + if directive == ConditionDirective.REMOVE: + # FDv1 failed permanently; FDv2 is still off until the TTL expires. + self._stop_event.wait(max(0.0, deadline - time.monotonic())) + return + def _consume_synchronizer_results( self, synchronizer: Synchronizer, set_on_ready: Event, check_recovery: bool, - ) -> ConditionDirective: + fdv2_recovery_ttl: Optional[float] = None, + ) -> Tuple[ConditionDirective, Optional[float]]: """ Consume results from a synchronizer until a condition is met or it fails. - :return: the ConditionDirective describing how to proceed + :param fdv2_recovery_ttl: when set, return RECOVER after this many seconds + :return: the ConditionDirective describing how to proceed, plus the + FDv1 fallback TTL in seconds when the directive is FDV1 """ action_queue: Queue = Queue() state_age = _StateAgeTracker() @@ -544,6 +594,10 @@ def _consume_synchronizer_results( initial_delay=10, callable=lambda: action_queue.put("check") ) + recovery_timer: Optional[Timer] = None + if fdv2_recovery_ttl is not None: + recovery_timer = Timer(fdv2_recovery_ttl, lambda: action_queue.put("recover")) + recovery_timer.daemon = True def reader(self: 'FDv2'): try: @@ -561,6 +615,8 @@ def reader(self: 'FDv2'): try: timer.start() + if recovery_timer is not None: + recovery_timer.start() sync_reader.start() while True: @@ -569,19 +625,22 @@ def reader(self: 'FDv2'): if update == "quit": break + if update == "recover": + return (ConditionDirective.RECOVER, None) + if update == "check": # Check condition periodically current_status = self._data_source_status_provider.status seconds_in_state = state_age.seconds_in_state(current_status) if check_recovery and recovery_condition(current_status, seconds_in_state): - return ConditionDirective.RECOVER + return (ConditionDirective.RECOVER, None) if fallback_condition(current_status, seconds_in_state): - return ConditionDirective.FALLBACK + return (ConditionDirective.FALLBACK, None) continue log.info("Synchronizer %s update: %s", synchronizer.name, update.state) if self._stop_event.is_set(): - return ConditionDirective.FALLBACK + return (ConditionDirective.FALLBACK, None) if update.state == DataSourceState.VALID: self._record_environment_id(update.environment_id) @@ -602,24 +661,27 @@ def reader(self: 'FDv2'): # response), in which case the ChangeSet has already been applied # above before we hand off. if update.fallback_to_fdv1: - return ConditionDirective.FDV1 + fallback_ttl = update.fallback_ttl if update.fallback_ttl is not None else fdv1_fallback_ttl() + return (ConditionDirective.FDV1, fallback_ttl) # Check for OFF state indicating permanent failure if update.state == DataSourceState.OFF: - return ConditionDirective.REMOVE + return (ConditionDirective.REMOVE, None) except Exception as e: log.error("Error consuming synchronizer results: %s", e) - return ConditionDirective.REMOVE + return (ConditionDirective.REMOVE, None) finally: synchronizer.stop() timer.stop() + if recovery_timer is not None: + recovery_timer.cancel() sync_reader.join(0.5) # If we reach here, the synchronizer's iterator completed normally (no more updates) # For continuous synchronizers (streaming/polling), this is unexpected and indicates # the synchronizer can't provide more updates, so we should remove it and fall back - return ConditionDirective.REMOVE + return (ConditionDirective.REMOVE, None) def _record_environment_id(self, environment_id: Optional[str]): if not isinstance(environment_id, str) or environment_id == '': diff --git a/ldclient/impl/datasystem/protocolv2.py b/ldclient/impl/datasystem/protocolv2.py index 7741a216..7b999145 100644 --- a/ldclient/impl/datasystem/protocolv2.py +++ b/ldclient/impl/datasystem/protocolv2.py @@ -4,6 +4,7 @@ """ from dataclasses import dataclass +from typing import Any, Dict, Optional from ldclient.interfaces import EventName, ObjectKind @@ -102,14 +103,22 @@ class Goodbye: """ reason: str + protocol_fallback_ttl: Optional[Any] = None + """ + The raw protocolFallbackTTL value, in seconds. When present, the server is + directing the SDK to fall back to FDv1. + """ def to_dict(self) -> dict: """ Serializes the Goodbye to a JSON-compatible dictionary. """ - return { + data: Dict[str, Any] = { "reason": self.reason, } + if self.protocol_fallback_ttl is not None: + data["protocolFallbackTTL"] = self.protocol_fallback_ttl + return data @staticmethod def from_dict(data: dict) -> "Goodbye": @@ -121,7 +130,7 @@ def from_dict(data: dict) -> "Goodbye": if reason is None: raise ValueError("Missing required fields in Goodbye JSON.") - return Goodbye(reason=reason) + return Goodbye(reason=reason, protocol_fallback_ttl=data.get("protocolFallbackTTL")) @dataclass(frozen=True) diff --git a/ldclient/impl/util.py b/ldclient/impl/util.py index deec78c8..0e4e8566 100644 --- a/ldclient/impl/util.py +++ b/ldclient/impl/util.py @@ -1,4 +1,5 @@ import logging +import random import re import sys import time @@ -44,6 +45,35 @@ def timedelta_millis(delta: timedelta) -> float: _LD_ENVID_HEADER = 'X-LD-EnvID' _LD_FD_FALLBACK_HEADER = 'X-LD-FD-Fallback' +_LD_FD_FALLBACK_TTL_HEADER = 'X-LD-FD-Fallback-TTL' + +_FDV1_FALLBACK_DEFAULT_TTL = 3600.0 + + +def fdv1_fallback_ttl(ttl: Any = None) -> float: + """ + Returns how many seconds the SDK stays off FDv2 after an FDv1 fallback directive. + + ``ttl`` is the service-supplied TTL in seconds, from the X-LD-FD-Fallback-TTL header or + a goodbye's protocolFallbackTTL. A TTL in (0, 1 hour] is used as given. Anything else, + including a missing or malformed TTL, gives the 1 hour default minus a random jitter of + up to 30 minutes, so that SDKs which fell back together do not all retry at once. + """ + if isinstance(ttl, str) and re.fullmatch('[0-9]+', ttl): + ttl = int(ttl) + if isinstance(ttl, (int, float)) and not isinstance(ttl, bool) and 0 < ttl <= _FDV1_FALLBACK_DEFAULT_TTL: + return float(ttl) + return _FDV1_FALLBACK_DEFAULT_TTL - random.uniform(0, _FDV1_FALLBACK_DEFAULT_TTL / 2) + + +def fdv1_fallback_ttl_from_headers(headers: Any) -> Optional[float]: + """ + Returns the FDv1 fallback TTL directed by a response's X-LD-FD-Fallback and + X-LD-FD-Fallback-TTL headers, or None when the headers carry no directive. + """ + if headers is None or headers.get(_LD_FD_FALLBACK_HEADER) != 'true': + return None + return fdv1_fallback_ttl(headers.get(_LD_FD_FALLBACK_TTL_HEADER)) def validate_application_info(application: dict, logger: logging.Logger) -> dict: diff --git a/ldclient/interfaces.py b/ldclient/interfaces.py index de403354..99ea39c6 100644 --- a/ldclient/interfaces.py +++ b/ldclient/interfaces.py @@ -1837,7 +1837,13 @@ class Basis: 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. + and then switch to the FDv1 Fallback Synchronizer until + ``fallback_ttl`` expires. + """ + fallback_ttl: Optional[float] = None + """ + How many seconds to stay off FDv2 when ``fallback_to_fdv1`` is True. + When None, the SDK uses its default. """ @@ -2088,11 +2094,17 @@ class Update: """ 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. + response header or a goodbye event). When True, callers must apply any + accompanying ``change_set`` first and then switch to the FDv1 Fallback + Synchronizer until ``fallback_ttl`` expires. The flag may ride along on + Valid, Interrupted, or Off updates. """ environment_id: Optional[str] = None + fallback_ttl: Optional[float] = None + """ + How many seconds to stay off FDv2 when ``fallback_to_fdv1`` is True. + When None, the SDK uses its default. + """ class Synchronizer(Protocol): # pylint: disable=too-few-public-methods diff --git a/ldclient/testing/impl/datasourcev2/test_async_streaming.py b/ldclient/testing/impl/datasourcev2/test_async_streaming.py index b3c55f02..ed974f30 100644 --- a/ldclient/testing/impl/datasourcev2/test_async_streaming.py +++ b/ldclient/testing/impl/datasourcev2/test_async_streaming.py @@ -23,7 +23,11 @@ Goodbye, PutObject ) -from ldclient.impl.util import _LD_ENVID_HEADER, _LD_FD_FALLBACK_HEADER +from ldclient.impl.util import ( + _LD_ENVID_HEADER, + _LD_FD_FALLBACK_HEADER, + _LD_FD_FALLBACK_TTL_HEADER +) from ldclient.interfaces import ( ChangeType, DataSourceErrorKind, @@ -549,3 +553,38 @@ async def test_envid_preserved_across_headerless_reconnect_start(): assert len(updates) == 2 assert updates[0].environment_id == 'my-env' assert updates[1].environment_id == 'my-env' + + +@pytest.mark.asyncio +async def test_goodbye_with_fallback_ttl_halts_with_signal(): + src = make_streaming_data_source() + + goodbye = Goodbye(reason="fall back", protocol_fallback_ttl=30) + actions = [ + Event(event=EventName.GOODBYE, data=json.dumps(goodbye.to_dict())), + # Should never be reached: the goodbye halts the stream. + server_intent_event(IntentCode.TRANSFER_FULL), + payload_transferred_event(), + ] + + updates = await collect_updates(src, actions) + assert len(updates) == 1 + assert updates[0].state == DataSourceState.INTERRUPTED + assert updates[0].fallback_to_fdv1 is True + assert updates[0].fallback_ttl == 30 + + +@pytest.mark.asyncio +async def test_fallback_header_ttl_is_carried_on_update(): + src = make_streaming_data_source() + + actions = [ + Start(headers={_LD_FD_FALLBACK_HEADER: 'true', _LD_FD_FALLBACK_TTL_HEADER: '120'}), + server_intent_event(IntentCode.TRANSFER_FULL), + payload_transferred_event(), + ] + + updates = await collect_updates(src, actions) + assert len(updates) == 1 + assert updates[0].fallback_to_fdv1 is True + assert updates[0].fallback_ttl == 120 diff --git a/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py b/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py index 8e4c6369..19dfd270 100644 --- a/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py +++ b/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py @@ -4,7 +4,13 @@ fdv1_polling_payload_to_changeset, polling_payload_to_changeset ) -from ldclient.impl.util import _Fail, _Success +from ldclient.impl.datasourcev2.polling_common import with_goodbye_fallback +from ldclient.impl.util import ( + _LD_ENVID_HEADER, + _Fail, + _Success, + fdv1_fallback_ttl_from_headers +) from ldclient.interfaces import ChangeType, IntentCode, ObjectKind @@ -359,3 +365,36 @@ def test_fdv1_payload_only_segments_no_flags_key(): change_set = result.value assert len(change_set.changes) == 1 assert change_set.changes[0].key == "test-segment" + + +def test_payload_with_only_goodbye_has_no_changes(): + data = {"events": [{"event": "goodbye", "data": {"reason": "fall back", "protocolFallbackTTL": 30}}]} + + result = polling_payload_to_changeset(data) + + assert isinstance(result, _Success) + assert result.value.intent_code == IntentCode.TRANSFER_NONE + + +def test_goodbye_fallback_ttl_is_added_to_headers(): + data = {"events": [{"event": "goodbye", "data": {"reason": "fall back", "protocolFallbackTTL": 30}}]} + + headers = with_goodbye_fallback({_LD_ENVID_HEADER: 'test-env'}, data) + + assert fdv1_fallback_ttl_from_headers(headers) == 30 + assert headers.get(_LD_ENVID_HEADER.lower()) == 'test-env' + + +def test_goodbye_with_invalid_fallback_ttl_uses_default(): + data = {"events": [{"event": "goodbye", "data": {"reason": "fall back", "protocolFallbackTTL": 7200}}]} + + headers = with_goodbye_fallback({}, data) + + assert 1800 <= fdv1_fallback_ttl_from_headers(headers) <= 3600 # type: ignore[operator] + + +def test_goodbye_without_fallback_ttl_leaves_headers_unchanged(): + data = {"events": [{"event": "goodbye", "data": {"reason": "bye"}}]} + headers = {_LD_ENVID_HEADER: 'test-env'} + + assert with_goodbye_fallback(headers, data) is headers diff --git a/ldclient/testing/impl/datasourcev2/test_polling_synchronizer.py b/ldclient/testing/impl/datasourcev2/test_polling_synchronizer.py index 925bc318..4c903234 100644 --- a/ldclient/testing/impl/datasourcev2/test_polling_synchronizer.py +++ b/ldclient/testing/impl/datasourcev2/test_polling_synchronizer.py @@ -15,6 +15,7 @@ from ldclient.impl.util import ( _LD_ENVID_HEADER, _LD_FD_FALLBACK_HEADER, + _LD_FD_FALLBACK_TTL_HEADER, UnsuccessfulResponseException, _Fail, _Success @@ -502,3 +503,35 @@ def test_synchronizer_generic_error_with_fallback_header_halts(): assert updates[0].environment_id == 'test-env-generic' assert updates[0].error is not None assert updates[0].error.kind == DataSourceErrorKind.NETWORK_ERROR + + +def test_fallback_header_ttl_is_carried_on_update(): + headers = {_LD_FD_FALLBACK_HEADER: 'true', _LD_FD_FALLBACK_TTL_HEADER: '45'} + polling_result: PollingResult = _Success(value=(ChangeSetBuilder.no_changes(), headers)) + + synchronizer = PollingDataSource( + poll_interval=0.01, requester=ListBasedRequester(results=iter([polling_result])) + ) + + valid = next(synchronizer.sync(MockSelectorStore(Selector.no_selector()))) + + assert valid.fallback_to_fdv1 is True + assert valid.fallback_ttl == 45 + + +def test_error_fallback_header_ttl_is_carried_on_update(): + headers = {_LD_FD_FALLBACK_HEADER: 'true', _LD_FD_FALLBACK_TTL_HEADER: '45'} + failure = _Fail( + error="error for test", + exception=UnsuccessfulResponseException(status=500), + headers=headers, + ) + + synchronizer = PollingDataSource( + poll_interval=0.01, requester=ListBasedRequester(results=iter([failure])) + ) + + update = next(synchronizer.sync(MockSelectorStore(Selector.no_selector()))) + + assert update.fallback_to_fdv1 is True + assert update.fallback_ttl == 45 diff --git a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py index 216013b4..adff6aa7 100644 --- a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py +++ b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py @@ -23,7 +23,11 @@ Goodbye, PutObject ) -from ldclient.impl.util import _LD_ENVID_HEADER, _LD_FD_FALLBACK_HEADER +from ldclient.impl.util import ( + _LD_ENVID_HEADER, + _LD_FD_FALLBACK_HEADER, + _LD_FD_FALLBACK_TTL_HEADER +) from ldclient.interfaces import ( ChangeType, DataSourceErrorKind, @@ -805,3 +809,63 @@ def test_envid_missing_when_no_headers(): assert len(updates) == 1 assert updates[0].state == DataSourceState.VALID assert updates[0].environment_id is None + + +def test_fallback_header_ttl_is_carried_on_update(events): # pylint: disable=redefined-outer-name + start_action = Start(headers={_LD_FD_FALLBACK_HEADER: 'true', _LD_FD_FALLBACK_TTL_HEADER: '120'}) + + builder = list_sse_client( + [ + start_action, + events[EventName.SERVER_INTENT], + events[EventName.PAYLOAD_TRANSFERRED], + ] + ) + + 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 + assert updates[0].fallback_ttl == 120 + + +def test_goodbye_with_fallback_ttl_halts_with_signal(events): # pylint: disable=redefined-outer-name + goodbye = Goodbye(reason="fall back", protocol_fallback_ttl=30) + goodbye_event = Event(event=EventName.GOODBYE, data=json.dumps(goodbye.to_dict())) + + builder = list_sse_client( + [ + Start(headers={_LD_ENVID_HEADER: 'test-env'}), + goodbye_event, + events[EventName.SERVER_INTENT], + events[EventName.PAYLOAD_TRANSFERRED], + ] + ) + + 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.INTERRUPTED + assert updates[0].fallback_to_fdv1 is True + assert updates[0].fallback_ttl == 30 + assert updates[0].environment_id == 'test-env' + assert updates[0].change_set is None + + +def test_goodbye_with_invalid_fallback_ttl_uses_default(): + goodbye = Goodbye(reason="fall back", protocol_fallback_ttl=0) + goodbye_event = Event(event=EventName.GOODBYE, data=json.dumps(goodbye.to_dict())) + + builder = list_sse_client([goodbye_event]) + + 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 + assert 1800 <= updates[0].fallback_ttl <= 3600 # type: ignore[operator] diff --git a/ldclient/testing/impl/datasystem/test_async_fdv2.py b/ldclient/testing/impl/datasystem/test_async_fdv2.py index 1a0a2d88..51308399 100644 --- a/ldclient/testing/impl/datasystem/test_async_fdv2.py +++ b/ldclient/testing/impl/datasystem/test_async_fdv2.py @@ -369,6 +369,7 @@ def __init__(self, name: str, updates: List[Update]): self._name = name self._updates = updates self.sync_called = False + self.sync_started = asyncio.Event() @property def name(self) -> str: @@ -376,6 +377,7 @@ def name(self) -> str: async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: self.sync_called = True + self.sync_started.set() for update in self._updates: yield update @@ -471,16 +473,26 @@ async def test_async_fdv2_initializer_header_fallback_engages_fdv1(): await fdv2.stop() +class SequenceAsyncSynchronizerBuilder(DataSourceBuilder): + """Builds each of the given synchronizers in turn.""" + + def __init__(self, synchronizers: List[Any]): + self._synchronizers = list(synchronizers) + + def build(self, config: DataSourceBuilderConfig): + return self._synchronizers.pop(0) + + @pytest.mark.asyncio -async def test_async_fdv2_initializer_header_fallback_without_fdv1_transitions_to_off(): +async def test_async_fdv2_initializer_fallback_without_fdv1_recovers_after_ttl(): """An initializer signals FDv1 fallback but no FDv1 synchronizer is - configured -> the data source transitions to OFF.""" - init = RecordingAsyncInitializer( - "hdr-fallback-no-fdv1", - _Fail(error="boom", exception=None, headers={_LD_FD_FALLBACK_HEADER: 'true'}), - ) + configured -> the SDK becomes ready and starts FDv2 after the TTL.""" + basis = _make_valid_basis() + basis = Basis(change_set=basis.change_set, persist=False, fallback_to_fdv1=True, fallback_ttl=0.05) - fdv2_sync = ListAsyncSynchronizer("fdv2-should-not-run", []) + init = RecordingAsyncInitializer("fallback-no-fdv1", _Success(value=basis)) + + fdv2_sync = ListAsyncSynchronizer("fdv2", []) data_system_config = DataSystemConfig( initializers=[MockAsyncInitializerBuilder(init)], # type: ignore[arg-type] @@ -493,10 +505,59 @@ async def test_async_fdv2_initializer_header_fallback_without_fdv1_transitions_t fdv2.start(ready_event) await asyncio.wait_for(ready_event.wait(), timeout=2) - await _status_reaches(fdv2, DataSourceState.OFF) + await asyncio.wait_for(fdv2_sync.sync_started.wait(), timeout=2) - assert fdv2.data_source_status_provider.status.state == DataSourceState.OFF - assert fdv2_sync.sync_called is False + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_synchronizer_fallback_without_fdv1_recovers_after_ttl(): + """A synchronizer signals FDv1 fallback with no FDv1 synchronizer + configured -> FDv2 is retried once the TTL expires.""" + fallback_update = Update(state=DataSourceState.INTERRUPTED, fallback_to_fdv1=True, fallback_ttl=0.05) + first = ListAsyncSynchronizer("first", [fallback_update]) + + second = ListAsyncSynchronizer("second", []) + + data_system_config = DataSystemConfig( + initializers=None, + synchronizers=[SequenceAsyncSynchronizerBuilder([first, second])], # type: ignore[list-item] + fdv1_fallback_synchronizer=None, + ) + + ready_event = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + fdv2.start(ready_event) + + await asyncio.wait_for(ready_event.wait(), timeout=2) + await asyncio.wait_for(second.sync_started.wait(), timeout=2) + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_returns_to_fdv2_after_fallback_ttl(): + """FDv1 serves data during the fallback, then FDv2 runs again after the TTL.""" + td_fdv1 = TestDataV2.data_source() + td_fdv1.update(td_fdv1.flag("fdv1-flag").on(True)) + + fallback_update = Update(state=DataSourceState.INTERRUPTED, fallback_to_fdv1=True, fallback_ttl=0.05) + first = ListAsyncSynchronizer("first", [fallback_update]) + + second = ListAsyncSynchronizer("second", []) + + data_system_config = DataSystemConfig( + initializers=None, + synchronizers=[SequenceAsyncSynchronizerBuilder([first, second])], # type: ignore[list-item] + fdv1_fallback_synchronizer=td_fdv1.async_builder, + ) + + ready_event = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + fdv2.start(ready_event) + + await asyncio.wait_for(second.sync_started.wait(), timeout=2) + assert await fdv2.store.get(FEATURES, "fdv1-flag") is not None await fdv2.stop() diff --git a/ldclient/testing/impl/datasystem/test_fdv2_datasystem.py b/ldclient/testing/impl/datasystem/test_fdv2_datasystem.py index 6cc98a85..795e17d1 100644 --- a/ldclient/testing/impl/datasystem/test_fdv2_datasystem.py +++ b/ldclient/testing/impl/datasystem/test_fdv2_datasystem.py @@ -4,7 +4,7 @@ import os import tempfile from threading import Event -from typing import List +from typing import List, Optional from mock import Mock @@ -439,24 +439,26 @@ def test_fdv2_no_fallback_without_header(): mock_secondary.sync.assert_called() -def test_fdv2_stays_on_fdv1_after_fallback(): +def test_fdv2_returns_to_fdv2_after_fallback_ttl(): """ - Test that once FDv2 falls back to FDv1, it stays on FDv1 and doesn't - attempt to recover to FDv2. + After an FDv1 fallback, FDv2 is stopped but retained: once the + server-supplied TTL expires, the FDv2 synchronizer runs again. """ - # Create mock primary that signals fallback + fallback_update = Update(state=DataSourceState.INTERRUPTED, fallback_to_fdv1=True, fallback_ttl=0.05) + + fdv2_restarted = Event() + + def fdv2_sync(ss): # pylint: disable=unused-argument + if mock_primary.sync.call_count == 1: + return iter([fallback_update]) + fdv2_restarted.set() + return iter([]) + mock_primary: Synchronizer = Mock() mock_primary.name = "mock-primary" mock_primary.stop = Mock() + mock_primary.sync.side_effect = fdv2_sync - mock_primary.sync.return_value = iter([ - Update( - state=DataSourceState.OFF, - fallback_to_fdv1=True - ) - ]) - - # Create FDv1 fallback td_fdv1 = TestDataV2.data_source() td_fdv1.update(td_fdv1.flag("fdv1-flag").on(True)) @@ -466,23 +468,57 @@ def test_fdv2_stays_on_fdv1_after_fallback(): fdv1_fallback_synchronizer=td_fdv1.builder, ) + fdv1_flag_seen = Event() + + def listener(flag_change: FlagChange): + if flag_change.key == "fdv1-flag": + fdv1_flag_seen.set() + set_on_ready = Event() fdv2 = FDv2(Config(sdk_key="dummy"), data_system_config) + fdv2.flag_change_listeners.add(listener) fdv2.start(set_on_ready) - assert set_on_ready.wait(1), "Data system did not become ready in time" + assert fdv1_flag_seen.wait(1), "FDv1 fallback synchronizer did not run" + assert fdv2_restarted.wait(1), "FDv2 synchronizer was not restarted after the TTL" - # Give it time to settle - import time - time.sleep(0.5) + fdv2.stop() + + +def test_fdv2_synchronizer_fallback_without_fdv1_configured_recovers_after_ttl(): + """ + When a synchronizer signals FDv1 fallback and no FDv1 Fallback + Synchronizer is configured, FDv2 is still retried once the TTL expires. + """ + fallback_update = Update(state=DataSourceState.INTERRUPTED, fallback_to_fdv1=True, fallback_ttl=0.05) + + fdv2_restarted = Event() + + def fdv2_sync(ss): # pylint: disable=unused-argument + if mock_primary.sync.call_count == 1: + return iter([fallback_update]) + fdv2_restarted.set() + return iter([]) + + mock_primary: Synchronizer = Mock() + mock_primary.name = "mock-primary" + mock_primary.stop = Mock() + mock_primary.sync.side_effect = fdv2_sync + + data_system_config = DataSystemConfig( + initializers=None, + synchronizers=[MockDataSourceBuilder(mock_primary)], + fdv1_fallback_synchronizer=None, + ) - # Primary should only be called once (not retried after fallback) - assert mock_primary.sync.call_count == 1 + set_on_ready = Event() + fdv2 = FDv2(Config(sdk_key="dummy"), data_system_config) + fdv2.start(set_on_ready) - # Verify FDv1 is serving data - store = fdv2.store - flag = store.get(FEATURES, "fdv1-flag", lambda x: x) - assert flag is not None + assert set_on_ready.wait(1), "Data system did not become ready in time" + assert fdv2_restarted.wait(1), "FDv2 synchronizer was not restarted after the TTL" + + fdv2.stop() def test_fdv2_initializer_should_run_until_success(): @@ -595,7 +631,7 @@ def listener(_: FlagChange): os.remove(path) -def _basis_with_one_flag(flag_key: str, fallback_to_fdv1: bool) -> Basis: +def _basis_with_one_flag(flag_key: str, fallback_to_fdv1: bool, fallback_ttl: Optional[float] = None) -> Basis: """Builds a Basis containing a single flag, optionally carrying the FDv1 fallback signal.""" builder = ChangeSetBuilder() builder.start(IntentCode.TRANSFER_FULL) @@ -611,6 +647,7 @@ def _basis_with_one_flag(flag_key: str, fallback_to_fdv1: bool) -> Basis: persist=True, environment_id=None, fallback_to_fdv1=fallback_to_fdv1, + fallback_ttl=fallback_ttl, ) @@ -664,26 +701,27 @@ def listener(flag_change: FlagChange): assert fdv1_flag_seen.wait(1), "FDv1 fallback synchronizer did not run after directive" -def test_fdv2_initializer_fallback_without_fdv1_configured_transitions_to_off(): +def test_fdv2_initializer_fallback_without_fdv1_configured_recovers_after_ttl(): """ When an initializer signals FDv1 fallback but no FDv1 Fallback Synchronizer - is configured, the data source status must transition to OFF rather than - silently dropping the directive or stalling at INITIALIZING. + is configured, the SDK becomes ready and starts the FDv2 synchronizers + once the TTL expires, rather than transitioning to OFF. """ init = _StaticInitializer( "fallback-initializer-no-fdv1", - _Fail( - error="boom", - exception=None, - headers={_LD_FD_FALLBACK_HEADER: 'true'}, - ), + _Success(value=_basis_with_one_flag("init-flag", fallback_to_fdv1=True, fallback_ttl=0.05)), ) - # An FDv2 synchronizer that would otherwise be tried -- it must not run. + fdv2_started = Event() + + def fdv2_sync(ss): # pylint: disable=unused-argument + fdv2_started.set() + return iter([Update(state=DataSourceState.VALID)]) + fdv2_sync_mock: Synchronizer = Mock() - fdv2_sync_mock.name = "fdv2-sync-should-not-run" + fdv2_sync_mock.name = "fdv2-sync" fdv2_sync_mock.stop = Mock() - fdv2_sync_mock.sync.return_value = iter([]) + fdv2_sync_mock.sync.side_effect = fdv2_sync data_system_config = DataSystemConfig( initializers=[_InitializerBuilder(init)], @@ -691,20 +729,14 @@ def test_fdv2_initializer_fallback_without_fdv1_configured_transitions_to_off(): fdv1_fallback_synchronizer=None, ) - off = Event() - - def listener(status: DataSourceStatus): - if status.state == DataSourceState.OFF: - off.set() - set_on_ready = Event() fdv2 = FDv2(Config(sdk_key="dummy"), data_system_config) - fdv2.data_source_status_provider.add_listener(listener) fdv2.start(set_on_ready) assert set_on_ready.wait(1), "Data system did not become ready in time" - assert off.wait(1), "Data source did not transition to OFF after directive without fallback" - fdv2_sync_mock.sync.assert_not_called() + assert fdv2_started.wait(1), "FDv2 synchronizer did not start after the TTL" + + fdv2.stop() def test_fdv2_synchronizer_fallback_on_success_with_payload(): diff --git a/ldclient/testing/impl/test_util.py b/ldclient/testing/impl/test_util.py index ea843e4b..592462f5 100644 --- a/ldclient/testing/impl/test_util.py +++ b/ldclient/testing/impl/test_util.py @@ -1,6 +1,16 @@ import logging -from ldclient.impl.util import validate_sdk_key_format +import pytest +from mock import patch + +from ldclient.impl import util +from ldclient.impl.util import ( + _LD_FD_FALLBACK_HEADER, + _LD_FD_FALLBACK_TTL_HEADER, + fdv1_fallback_ttl, + fdv1_fallback_ttl_from_headers, + validate_sdk_key_format +) def test_validate_sdk_key_format_valid(): @@ -63,3 +73,35 @@ def test_validate_sdk_key_format_max_length(): invalid_key = "a" * 8193 result = validate_sdk_key_format(invalid_key, logger) assert result == '' # Should return empty string for keys that are too long + + +@pytest.mark.parametrize("ttl, expected", [ + (1, 1.0), + (0.5, 0.5), + (3600, 3600.0), + ("1", 1.0), + ("3600", 3600.0), +]) +def test_fdv1_fallback_ttl_uses_valid_ttl_as_is(ttl, expected): + assert fdv1_fallback_ttl(ttl) == expected + + +@pytest.mark.parametrize("ttl", [None, 0, -1, 3601, "0", "-1", "3601", "", "abc", "1.5", True, {}]) +def test_fdv1_fallback_ttl_jitters_default_for_invalid_ttl(ttl): + assert 1800 <= fdv1_fallback_ttl(ttl) <= 3600 + + +@pytest.mark.parametrize("jitter, expected", [(0, 3600.0), (1800, 1800.0)]) +def test_fdv1_fallback_ttl_default_subtracts_up_to_half_an_hour(jitter, expected): + with patch.object(util.random, "uniform", return_value=jitter) as uniform: + assert fdv1_fallback_ttl(None) == expected + + uniform.assert_called_once_with(0, 1800) + + +def test_fdv1_fallback_ttl_from_headers(): + assert fdv1_fallback_ttl_from_headers({_LD_FD_FALLBACK_HEADER: 'true', _LD_FD_FALLBACK_TTL_HEADER: '30'}) == 30 + assert 1800 <= fdv1_fallback_ttl_from_headers({_LD_FD_FALLBACK_HEADER: 'true'}) <= 3600 # type: ignore[operator] + assert fdv1_fallback_ttl_from_headers({_LD_FD_FALLBACK_TTL_HEADER: '30'}) is None + assert fdv1_fallback_ttl_from_headers({}) is None + assert fdv1_fallback_ttl_from_headers(None) is None