From fc35432dc86a2ffc4728c96b5993931212b456cb Mon Sep 17 00:00:00 2001 From: "ayush.jain@juspay.in" Date: Tue, 14 Jul 2026 19:51:37 +0530 Subject: [PATCH] fix(openfeature): Python provider improvements --- clients/python/provider-sdk-tests/main.py | 4 +- .../superposition_provider/__init__.py | 11 +- .../superposition_provider/cac_config.py | 15 +- .../superposition_provider/data_source.py | 30 ++- .../provider/superposition_provider/errors.py | 61 +++++ .../superposition_provider/exp_config.py | 15 +- .../file_data_source.py | 151 ++++++++---- .../http_data_source.py | 47 +++- .../superposition_provider/interfaces.py | 60 +++-- .../superposition_provider/local_provider.py | 221 ++++++++++++++---- .../superposition_provider/provider.py | 3 - .../superposition_provider/remote_provider.py | 8 +- .../provider/superposition_provider/types.py | 114 ++++++--- clients/python/provider/test_file_watch.py | 116 +++++++++ .../python/provider/test_http_data_source.py | 116 +++++++++ clients/python/provider/test_ondemand_ttl.py | 135 +++++++++++ clients/python/provider/test_type_contract.py | 103 ++++++++ 17 files changed, 1024 insertions(+), 186 deletions(-) create mode 100644 clients/python/provider/superposition_provider/errors.py create mode 100644 clients/python/provider/test_file_watch.py create mode 100644 clients/python/provider/test_http_data_source.py create mode 100644 clients/python/provider/test_ondemand_ttl.py create mode 100644 clients/python/provider/test_type_contract.py diff --git a/clients/python/provider-sdk-tests/main.py b/clients/python/provider-sdk-tests/main.py index 1a9f46375..ef1874849 100644 --- a/clients/python/provider-sdk-tests/main.py +++ b/clients/python/provider-sdk-tests/main.py @@ -314,8 +314,8 @@ async def setup_with_sdk(client, org_id: str, workspace_id: str): async def run_demo(org_id: str, workspace_id: str): refresh_strategy = PollingStrategy( - interval=5, - timeout=3, + interval_milliseconds=5_000, + timeout_milliseconds=3_000, ) http_options = SuperpositionOptions( endpoint="http://localhost:8080", diff --git a/clients/python/provider/superposition_provider/__init__.py b/clients/python/provider/superposition_provider/__init__.py index 3ee99e8c4..55e82d85a 100644 --- a/clients/python/provider/superposition_provider/__init__.py +++ b/clients/python/provider/superposition_provider/__init__.py @@ -23,9 +23,10 @@ from .local_provider import LocalResolutionProvider, RefreshStrategy from .remote_provider import SuperpositionAPIProvider +from .errors import SuperpositionError, ErrorCode + from .types import ( SuperpositionOptions, - EvaluationCacheOptions, PollingStrategy, OnDemandStrategy, WatchStrategy, @@ -33,8 +34,6 @@ RefreshStrategy as RefreshStrategyType, ConfigurationOptions, ExperimentationOptions, - LocalProviderOptions, - RemoteProviderOptions, SuperpositionProviderOptions, ) @@ -53,9 +52,11 @@ "LocalResolutionProvider", "SuperpositionAPIProvider", "RefreshStrategy", + # Errors + "SuperpositionError", + "ErrorCode", # Types "SuperpositionOptions", - "EvaluationCacheOptions", "PollingStrategy", "OnDemandStrategy", "WatchStrategy", @@ -63,7 +64,5 @@ "RefreshStrategyType", "ConfigurationOptions", "ExperimentationOptions", - "LocalProviderOptions", - "RemoteProviderOptions", "SuperpositionProviderOptions", ] diff --git a/clients/python/provider/superposition_provider/cac_config.py b/clients/python/provider/superposition_provider/cac_config.py index 6b27697e1..379b3817c 100644 --- a/clients/python/provider/superposition_provider/cac_config.py +++ b/clients/python/provider/superposition_provider/cac_config.py @@ -121,7 +121,7 @@ async def poll_config(interval: int, timeout: int) -> None: finally: del self_ref - await asyncio.sleep(interval) + await asyncio.sleep(interval / 1000) # interval is in milliseconds latest_config = await self._get_config(self.superposition_options) if latest_config is None and self.cached_config is None: @@ -140,13 +140,18 @@ async def poll_config(interval: int, timeout: int) -> None: self.on_config_change() match self.options.refresh_strategy: - case PollingStrategy(interval=interval, timeout=timeout): - logger.info(f"Using PollingStrategy: interval={interval}, timeout={timeout}") + case PollingStrategy(): + interval = self.options.refresh_strategy.interval_ms() + timeout = self.options.refresh_strategy.timeout_ms() + logger.info(f"Using PollingStrategy: interval={interval}ms, timeout={timeout}ms") if self._polling_task is None: self._polling_task = asyncio.create_task(poll_config(interval, timeout)) - case OnDemandStrategy(ttl=ttl, use_stale_on_error=use_stale, timeout=timeout): - logger.info(f"Using OnDemandStrategy: ttl={ttl}, use_stale_on_error={use_stale}, timeout={timeout}") + case OnDemandStrategy(): + ttl = self.options.refresh_strategy.ttl_ms() + use_stale = self.options.refresh_strategy.use_stale_on_error + timeout = self.options.refresh_strategy.timeout_ms() + logger.info(f"Using OnDemandStrategy: ttl={ttl}ms, use_stale_on_error={use_stale}, timeout={timeout}ms") diff --git a/clients/python/provider/superposition_provider/data_source.py b/clients/python/provider/superposition_provider/data_source.py index 263b177f9..eeb683b21 100644 --- a/clients/python/provider/superposition_provider/data_source.py +++ b/clients/python/provider/superposition_provider/data_source.py @@ -19,29 +19,45 @@ T = TypeVar('T') class FetchResponse(Generic[T]): - """Represents a fetch response with optional data, supporting 304 Not Modified.""" + """Either fetched data or a 304 Not Modified marker. - def __init__(self, data: Optional[T] = None): + A true sum type: `NotModified` is a distinct variant rather than "data is None", so a data + source that legitimately returns nothing cannot be mistaken for an unchanged one. + """ + + __slots__ = ("_data", "_not_modified") + + def __init__(self, data: Optional[T] = None, not_modified: bool = False): self._data = data + self._not_modified = not_modified @staticmethod - def not_modified(): + def not_modified() -> "FetchResponse[T]": """Create a 304 Not Modified response.""" - return FetchResponse(data=None) + return FetchResponse(data=None, not_modified=True) @staticmethod - def data(data: T): + def data(data: T) -> "FetchResponse[T]": """Create a successful response with data.""" - return FetchResponse(data=data) + return FetchResponse(data=data, not_modified=False) def is_not_modified(self) -> bool: """Check if this is a 304 Not Modified response.""" - return self._data is None + return self._not_modified def get_data(self) -> Optional[T]: """Get the response data, or None if not modified.""" return self._data + def map_data(self, mapper): + """Transform the data if present, preserving a NotModified response.""" + if self._not_modified: + return FetchResponse.not_modified() + return FetchResponse.data(mapper(self._data)) + + def __repr__(self) -> str: + return "FetchResponse.NotModified" if self._not_modified else f"FetchResponse.Data({self._data})" + @dataclass class ConfigData: diff --git a/clients/python/provider/superposition_provider/errors.py b/clients/python/provider/superposition_provider/errors.py new file mode 100644 index 000000000..5fe988931 --- /dev/null +++ b/clients/python/provider/superposition_provider/errors.py @@ -0,0 +1,61 @@ +""" +Typed errors raised by the Superposition provider. + +Mirrors the Rust `SuperpositionError` enum and the Java `SuperpositionError` exception, so a +failure means the same thing in every client. +""" + +from enum import Enum +from typing import Optional + + +class ErrorCode(Enum): + """The kind of failure a SuperpositionError represents.""" + + CONFIG_ERROR = "CONFIG_ERROR" + NETWORK_ERROR = "NETWORK_ERROR" + SERIALIZATION_ERROR = "SERIALIZATION_ERROR" + PROVIDER_ERROR = "PROVIDER_ERROR" + DATA_SOURCE_ERROR = "DATA_SOURCE_ERROR" + REFRESH_ERROR = "REFRESH_ERROR" + + +class SuperpositionError(Exception): + """Raised when a provider or data source operation fails.""" + + def __init__(self, code: ErrorCode, message: str, cause: Optional[BaseException] = None): + super().__init__(message) + self.code = code + self.message = message + self.cause = cause + if cause is not None: + self.__cause__ = cause + + def __str__(self) -> str: + return f"{self.code.value}: {self.message}" + + # --- Factories, one per variant --- + + @staticmethod + def config_error(message: str, cause: Optional[BaseException] = None) -> "SuperpositionError": + return SuperpositionError(ErrorCode.CONFIG_ERROR, message, cause) + + @staticmethod + def network_error(message: str, cause: Optional[BaseException] = None) -> "SuperpositionError": + return SuperpositionError(ErrorCode.NETWORK_ERROR, message, cause) + + @staticmethod + def serialization_error(message: str, cause: Optional[BaseException] = None) -> "SuperpositionError": + return SuperpositionError(ErrorCode.SERIALIZATION_ERROR, message, cause) + + @staticmethod + def provider_error(message: str, cause: Optional[BaseException] = None) -> "SuperpositionError": + return SuperpositionError(ErrorCode.PROVIDER_ERROR, message, cause) + + @staticmethod + def data_source_error(message: str, cause: Optional[BaseException] = None) -> "SuperpositionError": + return SuperpositionError(ErrorCode.DATA_SOURCE_ERROR, message, cause) + + @staticmethod + def refresh_error(message: str, cause: Optional[BaseException] = None) -> "SuperpositionError": + return SuperpositionError(ErrorCode.REFRESH_ERROR, message, cause) diff --git a/clients/python/provider/superposition_provider/exp_config.py b/clients/python/provider/superposition_provider/exp_config.py index d4eaf8a7f..494209b14 100644 --- a/clients/python/provider/superposition_provider/exp_config.py +++ b/clients/python/provider/superposition_provider/exp_config.py @@ -76,7 +76,7 @@ async def poll_config(interval: int, timeout: int) -> None: finally: del self_ref - await asyncio.sleep(interval) + await asyncio.sleep(interval / 1000) # interval is in milliseconds latest_exp_list = await self._get_experiments(self.superposition_options) latest_exp_grp_list = await self._get_experiment_groups(self.superposition_options) @@ -92,13 +92,18 @@ async def poll_config(interval: int, timeout: int) -> None: self.on_config_change() match self.options.refresh_strategy: - case PollingStrategy(interval=interval, timeout=timeout): - logger.info(f"Using PollingStrategy: interval={interval}, timeout={timeout}") + case PollingStrategy(): + interval = self.options.refresh_strategy.interval_ms() + timeout = self.options.refresh_strategy.timeout_ms() + logger.info(f"Using PollingStrategy: interval={interval}ms, timeout={timeout}ms") if self._polling_task is None: self._polling_task = asyncio.create_task(poll_config(interval, timeout)) - case OnDemandStrategy(ttl=ttl, use_stale_on_error=use_stale, timeout=timeout): - logger.info(f"Using OnDemandStrategy: ttl={ttl}, use_stale_on_error={use_stale}, timeout={timeout}") + case OnDemandStrategy(): + ttl = self.options.refresh_strategy.ttl_ms() + use_stale = self.options.refresh_strategy.use_stale_on_error + timeout = self.options.refresh_strategy.timeout_ms() + logger.info(f"Using OnDemandStrategy: ttl={ttl}ms, use_stale_on_error={use_stale}, timeout={timeout}ms") @staticmethod async def _get_experiments(superposition_options: SuperpositionOptions) -> Optional[list[FfiExperiment]]: diff --git a/clients/python/provider/superposition_provider/file_data_source.py b/clients/python/provider/superposition_provider/file_data_source.py index 70a9f69fe..f7f130fed 100644 --- a/clients/python/provider/superposition_provider/file_data_source.py +++ b/clients/python/provider/superposition_provider/file_data_source.py @@ -17,6 +17,7 @@ from superposition_bindings.superposition_types import Config from watchdog.observers import Observer +from watchdog.observers.api import BaseObserver from watchdog.events import FileSystemEventHandler from .data_source import ( @@ -25,35 +26,68 @@ ConfigData, ExperimentData, ) +from .errors import SuperpositionError logger = logging.getLogger(__name__) + +def _stop_observer(observer: Optional[BaseObserver]) -> None: + """Release a watchdog Observer's OS handle, whether or not it ever started. + + The join is guarded. If setup failed before ``start()`` — ``schedule()`` raises OSError on + inotify when the directory is missing — then joining a thread that was never started raises + ``RuntimeError: cannot join thread before it is started``. Raised from a ``finally``, that + would *replace* the real error, and the caller would be told their thread is unstartable + rather than that their directory does not exist. + """ + if observer is None: + return + try: + observer.stop() + if observer.is_alive(): + observer.join(timeout=1.0) + except Exception as e: # cleanup must never mask the failure that triggered it + logger.warning(f"Error stopping file watcher: {e}") + + class _FileEventHandler(FileSystemEventHandler): - """Handler for file system events from watchdog.""" + """Fans every change to the watched file out to all live subscribers. - def __init__(self, file_path: str, loop: asyncio.AbstractEventLoop, queue: asyncio.Queue): + One handler serves every subscriber: a single OS watcher, N queues. + An Observer per subscriber would mean several watchers registered on the same + directory, which is both wasteful and unreliable. + """ + + def __init__( + self, + file_path: str, + loop: asyncio.AbstractEventLoop, + subscribers: List[asyncio.Queue], + ): """Initialize handler. Args: file_path: Path to the file to watch. loop: The asyncio event loop to schedule callbacks on. - queue: The asyncio queue to put change events into. + subscribers: Live subscriber queues; read on each event, so late joiners are served. """ super().__init__() self.file_path = file_path self._loop = loop - self._queue = queue + self._subscribers = subscribers def on_modified(self, event): """Called when a file is modified (from watchdog's background thread).""" if event.is_directory: return - # Check if the modified file matches our watched file - if os.path.abspath(event.src_path) == os.path.abspath(self.file_path): - logger.info(f"File changed (watchdog event): {self.file_path}") - # Thread-safe: schedule the queue put from the watchdog thread - self._loop.call_soon_threadsafe(self._queue.put_nowait, self.file_path) + if os.path.realpath(event.src_path) != os.path.realpath(self.file_path): + return + + logger.info(f"File changed (watchdog event): {self.file_path}") + # Thread-safe: schedule the queue puts from the watchdog thread. + for queue in list(self._subscribers): + self._loop.call_soon_threadsafe(queue.put_nowait, self.file_path) def _parse_config_file( content: str, @@ -91,14 +125,17 @@ def __init__( Args: file_path: Path to configuration TOML/JSON file. """ - self._watch_task = None + self._observer: Optional[BaseObserver] = None + self._subscribers: List[asyncio.Queue] = [] self.file_path = file_path if file_path.endswith('.toml'): self.file_format = "toml" elif file_path.endswith('.json'): self.file_format = "json" else: - raise ValueError(f"Unsupported file format: {file_path}") + raise SuperpositionError.data_source_error( + f"Unsupported file extension: {file_path}. Supported formats are 'json' and 'toml'." + ) async def fetch_filtered_config( self, @@ -134,9 +171,13 @@ async def fetch_filtered_config( data=config, fetched_at=now, )) + except SuperpositionError: + raise except Exception as e: logger.error(f"Failed to fetch config from {self.file_path}: {e}") - raise + raise SuperpositionError.data_source_error( + f"Failed to read config file {self.file_path}: {e}", e + ) from e async def fetch_active_experiments( self, @@ -150,7 +191,7 @@ async def fetch_active_experiments( Returns: FetchResponse with ExperimentData or NotModified status. """ - raise NotImplementedError("Experiments not supported by FileDataSource") + raise SuperpositionError.data_source_error("Experiments not supported by FileDataSource") async def fetch_candidate_active_experiments( self, @@ -159,7 +200,7 @@ async def fetch_candidate_active_experiments( if_modified_since: Optional[datetime] = None, ) -> FetchResponse[ExperimentData]: """Fetch candidate active experiments.""" - raise NotImplementedError("Experiments not supported by FileDataSource") + raise SuperpositionError.data_source_error("Experiments not supported by FileDataSource") async def fetch_matching_active_experiments( self, @@ -168,19 +209,17 @@ async def fetch_matching_active_experiments( if_modified_since: Optional[datetime] = None, ) -> FetchResponse[ExperimentData]: """Fetch matching active experiments.""" - raise NotImplementedError("Experiments not supported by FileDataSource") + raise SuperpositionError.data_source_error("Experiments not supported by FileDataSource") def supports_experiments(self) -> bool: """File source supports experiments if path is configured.""" return False async def watch(self) -> Optional[AsyncGenerator[str, None]]: - """Set up file watching for changes using native file system notifications. + """Watch the file for changes, yielding its path on every change. - Uses watchdog library which provides: - - inotify on Linux - - FSEvents on macOS - - ReadDirectoryChangesW on Windows + Every caller gets its own stream of events. They share one OS watcher, which starts with + the first subscriber and stops when the last one leaves. Returns: Async generator yielding changed file paths. @@ -188,47 +227,59 @@ async def watch(self) -> Optional[AsyncGenerator[str, None]]: if not self.file_path: return + event_queue: asyncio.Queue = asyncio.Queue() + self._subscribers.append(event_queue) try: - # Create a queue to bridge sync events to async - event_queue: asyncio.Queue = asyncio.Queue() - loop = asyncio.get_running_loop() - - # Set up watchdog observer - self._watch_task = Observer() - event_handler = _FileEventHandler(self.file_path, loop, event_queue) + self._ensure_observer() + logger.info(f"Watching file: {self.file_path} (subscribers: {len(self._subscribers)})") - # Watch the directory containing the file - watch_dir = os.path.dirname(os.path.abspath(self.file_path)) - self._watch_task.schedule(event_handler, watch_dir, recursive=False) - self._watch_task.start() - - logger.info(f"Watching file: {self.file_path} using inotify (watchdog)") - - # Yield events as they arrive while True: try: - # Wait for file change event (with timeout to allow cancellation) - changed_path = await asyncio.wait_for( - event_queue.get(), - timeout=5.0 - ) + # Time-limited so a cancelled consumer is noticed promptly. + changed_path = await asyncio.wait_for(event_queue.get(), timeout=5.0) yield changed_path except asyncio.TimeoutError: - # No events, continue watching continue except asyncio.CancelledError: logger.debug("File watch cancelled") break finally: - # Clean up observer - if self._watch_task: - self._watch_task.stop() - self._watch_task.join(timeout=1.0) - logger.debug("File watcher stopped") + if event_queue in self._subscribers: + self._subscribers.remove(event_queue) + # The watcher exists for the subscribers; with none left it is just an open handle. + if not self._subscribers: + observer, self._observer = self._observer, None + _stop_observer(observer) + logger.debug("Last subscriber left, file watcher stopped") + + def _ensure_observer(self) -> None: + """Start the shared watcher, unless a previous subscriber already did. + + The field is published only once ``start()`` has succeeded, so a failure here leaves + nothing half-registered for ``close()`` — or the next subscriber — to trip over. + """ + if self._observer is not None: + return + + loop = asyncio.get_running_loop() + handler = _FileEventHandler(self.file_path, loop, self._subscribers) + # realpath, so the directory we register matches the paths the OS reports back. + watch_dir = os.path.dirname(os.path.realpath(self.file_path)) + + observer = Observer() + try: + observer.schedule(handler, watch_dir, recursive=False) + observer.start() + except Exception as e: + _stop_observer(observer) + raise SuperpositionError.data_source_error( + f"Failed to watch {self.file_path}: {e}", e + ) from e + + self._observer = observer async def close(self) -> None: """Stop watching and clean up resources.""" - if self._watch_task: - self._watch_task.stop() - self._watch_task.join(timeout=1.0) - self._watch_task = None + observer, self._observer = self._observer, None + self._subscribers.clear() + _stop_observer(observer) diff --git a/clients/python/provider/superposition_provider/http_data_source.py b/clients/python/provider/superposition_provider/http_data_source.py index 7b6e3974b..ebc38343b 100644 --- a/clients/python/provider/superposition_provider/http_data_source.py +++ b/clients/python/provider/superposition_provider/http_data_source.py @@ -9,15 +9,17 @@ from typing import Dict, List, Optional, Any from smithy_core.documents import Document +from smithy_core.interceptors import Interceptor, ResponseContext from superposition_bindings.superposition_client import ExperimentConfig from superposition_sdk.client import Superposition from superposition_sdk.config import Config as SdkConfig from superposition_sdk.auth_helpers import bearer_auth_config from superposition_sdk.models import GetConfigInput, DimensionMatchStrategy, \ - GetExperimentConfigInput, UnknownApiError + GetExperimentConfigInput, ServiceError from .conversions import experiments_to_ffi_experiments, exp_grps_to_ffi_exp_grps, config_response_to_ffi_config +from .errors import SuperpositionError from .types import SuperpositionOptions from .data_source import ( SuperpositionDataSource, @@ -29,6 +31,34 @@ logger = logging.getLogger(__name__) +class _NotModified(ServiceError): + """The service answered 304: the caller's cached copy is still current. + + Deliberately a `ServiceError`: the client wraps anything else in one on its way out + (`raise ServiceError(e) from e`), which would bury this behind a cause chain. + """ + + +class _NotModifiedInterceptor(Interceptor): + """Turns an HTTP 304 into a distinguishable signal. + + The generated SDK models no shape for 304, and its deserializer treats every status + >= 300 as a failure, collapsing it into an `UnknownApiError` that carries only a + message string. Reading the raw status before deserialization is the only way to tell + "your cached copy is still current" from a genuine failure — the Java client solves + this the same way. + + This replaces an `except UnknownApiError: return not_modified()` catch-all, which + reported expired credentials and 500s as a successful "nothing changed" and silently + froze the config. + """ + + def read_after_transmit(self, context: ResponseContext) -> None: + response = context.transport_response + if response is not None and response.status == 304: + raise _NotModified() + + class HttpDataSource(SuperpositionDataSource): """HTTP-based data source for Superposition API. @@ -56,7 +86,8 @@ def _create_client(self) -> Superposition: sdk_config = SdkConfig( endpoint_uri=self.options.endpoint, http_auth_scheme_resolver=resolver, - http_auth_schemes=schemes + http_auth_schemes=schemes, + interceptors=[_NotModifiedInterceptor()], ) # Create Superposition client @@ -93,11 +124,10 @@ async def fetch_filtered_config( fetched_at=response.last_modified, data=config_response_to_ffi_config(response), )) - except UnknownApiError as e: - # this is a hack for now, need to fix it properly by checking the content of UnkownApiError + except _NotModified: return FetchResponse.not_modified() except Exception as e: - raise e + raise SuperpositionError.network_error(f"Failed to fetch config: {e}", e) from e async def _fetch_filtered_experiment( self, @@ -126,11 +156,12 @@ async def _fetch_filtered_experiment( experiment_groups=exp_grps_to_ffi_exp_grps(response.experiment_groups), ) )) - except UnknownApiError as e: - # this is a hack for now, need to fix it properly by checking the content of UnkownApiError + except _NotModified: return FetchResponse.not_modified() except Exception as e: - raise e + raise SuperpositionError.network_error( + f"Failed to fetch experiments: {e}", e + ) from e async def fetch_active_experiments( self, diff --git a/clients/python/provider/superposition_provider/interfaces.py b/clients/python/provider/superposition_provider/interfaces.py index 5ab383a8d..d61de6fe9 100644 --- a/clients/python/provider/superposition_provider/interfaces.py +++ b/clients/python/provider/superposition_provider/interfaces.py @@ -10,7 +10,7 @@ from openfeature.evaluation_context import EvaluationContext from openfeature.exception import ErrorCode -from openfeature.flag_evaluation import FlagResolutionDetails +from openfeature.flag_evaluation import FlagResolutionDetails, Reason logger = logging.getLogger(__name__) @@ -90,6 +90,13 @@ def resolve_typed( Resolves all features and extracts a specific flag, applying type conversion. + TODO: successful resolutions leave `reason` unset. Reporting it accurately (STATIC for a + default-config value, TARGETING_MATCH for a context override, SPLIT for an experiment + variant) needs `eval_config` in superposition_core to say, per key, where the value came + from. Until it does, guessing would be worse than saying nothing — a flag no experiment + touched would still be labelled SPLIT. Error reasons are set below. The same TODO applies + to the Rust and Java clients. + Args: flag_key: The flag key to resolve. evaluation_context: Evaluation context. @@ -103,17 +110,33 @@ def resolve_typed( try: config = self.resolve_all_features(evaluation_context) if flag_key not in config: - return FlagResolutionDetails(default) + return FlagResolutionDetails( + default, + reason=Reason.ERROR, + error_code=ErrorCode.FLAG_NOT_FOUND, + error_message=f"Flag '{flag_key}' not found" + ) value = config[flag_key] extracted = extractor(value) if extracted is None: - return FlagResolutionDetails(default) + return FlagResolutionDetails( + default, + reason=Reason.ERROR, + error_code=ErrorCode.TYPE_MISMATCH, + error_message=f"Flag '{flag_key}' is not a {type_name}" + ) return FlagResolutionDetails(extracted) + except NotImplementedError: + # A provider that cannot resolve this way at all (SuperpositionAPIProvider is + # async-only). Reporting it as a per-flag GENERAL error made every sync call quietly + # return the default forever, with nothing to distinguish it from a real evaluation. + raise except Exception as e: logger.error(f"Error evaluating {type_name} flag {flag_key}: {e}") return FlagResolutionDetails( default, + reason=Reason.ERROR, error_code=ErrorCode.GENERAL, error_message=f"Error evaluating flag '{flag_key}': {e}" ) @@ -129,7 +152,7 @@ def resolve_bool( flag_key, evaluation_context, "boolean", - lambda v: _to_bool(v), + lambda v: v if isinstance(v, bool) else None, default_value ) @@ -249,17 +272,33 @@ async def resolve_typed_async( try: config = await self.resolve_all_features_async(evaluation_context) if flag_key not in config: - return FlagResolutionDetails(default) + return FlagResolutionDetails( + default, + reason=Reason.ERROR, + error_code=ErrorCode.FLAG_NOT_FOUND, + error_message=f"Flag '{flag_key}' not found" + ) value = config[flag_key] extracted = extractor(value) if extracted is None: - return FlagResolutionDetails(default) + return FlagResolutionDetails( + default, + reason=Reason.ERROR, + error_code=ErrorCode.TYPE_MISMATCH, + error_message=f"Flag '{flag_key}' is not a {type_name}" + ) return FlagResolutionDetails(extracted) + except NotImplementedError: + # A provider that cannot resolve this way at all (SuperpositionAPIProvider is + # async-only). Reporting it as a per-flag GENERAL error made every sync call quietly + # return the default forever, with nothing to distinguish it from a real evaluation. + raise except Exception as e: logger.error(f"Error evaluating {type_name} flag {flag_key}: {e}") return FlagResolutionDetails( default, + reason=Reason.ERROR, error_code=ErrorCode.GENERAL, error_message=f"Error evaluating flag '{flag_key}': {e}" ) @@ -275,7 +314,7 @@ async def resolve_bool_async( flag_key, evaluation_context, "boolean", - lambda v: _to_bool(v), + lambda v: v if isinstance(v, bool) else None, default_value ) @@ -339,10 +378,3 @@ async def resolve_object_async( default_value ) -def _to_bool(value: Any) -> Optional[bool]: - if isinstance(value, bool): return value - if isinstance(value, str): - if value.lower() == "true": return True - if value.lower() == "false": return False - if isinstance(value, (int, float)): return value != 0 - return None diff --git a/clients/python/provider/superposition_provider/local_provider.py b/clients/python/provider/superposition_provider/local_provider.py index a8db676dc..63634dc50 100644 --- a/clients/python/provider/superposition_provider/local_provider.py +++ b/clients/python/provider/superposition_provider/local_provider.py @@ -18,6 +18,7 @@ ProviderStatus, ) from openfeature.evaluation_context import EvaluationContext +from openfeature.event import ProviderEventDetails from openfeature.flag_evaluation import FlagResolutionDetails from superposition_bindings.superposition_client import ProviderCache @@ -25,6 +26,7 @@ from . import FetchResponse from .data_source import SuperpositionDataSource, ConfigData, ExperimentData +from .errors import SuperpositionError from .interfaces import AllFeatureProvider, FeatureExperimentMeta from .types import RefreshStrategy, OnDemandStrategy, WatchStrategy, PollingStrategy, ManualStrategy, default_on_demand_strategy @@ -44,18 +46,20 @@ def __init__( self, primary_source: SuperpositionDataSource, fallback_source: Optional[SuperpositionDataSource] = None, - refresh_strategy: RefreshStrategy = default_on_demand_strategy(), + refresh_strategy: Optional[RefreshStrategy] = None, ): """Initialize local resolution provider. Args: primary_source: Primary data source for config/experiments. fallback_source: Optional fallback data source. - refresh_strategy: How often to refresh data. + refresh_strategy: How often to refresh data. Defaults to on-demand. """ self.primary_source = primary_source self.fallback_source = fallback_source - self.refresh_strategy = refresh_strategy + # Built per instance, not in the signature: a default argument is evaluated once at + # import, so every provider in the process would otherwise share one strategy object. + self.refresh_strategy = refresh_strategy or default_on_demand_strategy() self.metadata = Metadata(name="LocalResolutionProvider") self.status = ProviderStatus.NOT_READY @@ -66,6 +70,19 @@ def __init__( self.cached_experiments: Optional[ExperimentData] = None self.ffi_cache: Optional[ProviderCache] = None + # When each cache was last successfully checked against its source, by the local clock. + # + # Deliberately not ConfigData.fetched_at: for an HTTP source that is the *server's* + # last-modified — when the config last *changed*, not when we last *looked*. Driving the + # ON_DEMAND TTL off it meant a config stable for longer than the TTL was permanently + # "stale", so every evaluation fired a fetch; the 304 that came back left the timestamp + # untouched, so the next evaluation fired another. A perfectly stable config produced + # maximum load, which is the opposite of what ON_DEMAND is for. + # + # Advanced on every successful check, *including* a 304. + self.config_checked_at: Optional[datetime] = None + self.experiments_checked_at: Optional[datetime] = None + # Background task for refresh strategy self._background_task: Optional[asyncio.Task] = None @@ -77,6 +94,16 @@ async def initialize(self, context: EvaluationContext): Args: context: Global evaluation context. """ + # Single-shot: a provider is initialized once and then served. Re-initializing a live + # provider would overwrite the background-task handle, orphaning the running polling/watch + # loop, so it is refused. A fresh or previously-failed provider proceeds. + if self.status in (ProviderStatus.READY, ProviderStatus.STALE): + logger.warning( + "LocalResolutionProvider already initialized; ignoring initialize(). " + "Providers are single-shot — build a new instance." + ) + return + try: logger.info("Initializing LocalResolutionProvider...") self.status = ProviderStatus.NOT_READY @@ -98,7 +125,7 @@ async def initialize(self, context: EvaluationContext): logger.info("LocalResolutionProvider initialized successfully") except Exception as e: logger.error(f"Failed to initialize LocalResolutionProvider: {e}") - self.status = ProviderStatus.FATAL + self.status = ProviderStatus.ERROR raise async def shutdown(self): @@ -129,6 +156,8 @@ async def shutdown(self): self.cached_config = None self.cached_experiments = None self.ffi_cache = None + self.config_checked_at = None + self.experiments_checked_at = None self.status = ProviderStatus.NOT_READY logger.info("LocalResolutionProvider shutdown completed") @@ -162,7 +191,7 @@ def resolve_all_features_with_filter( case _: () if not self.ffi_cache or not self.cached_config: - raise RuntimeError("Provider not properly initialized") + raise SuperpositionError.provider_error("Provider not initialized: no cached config available") # Merge contexts targeting_key, query_data = self._merge_contexts(context) @@ -176,12 +205,28 @@ def resolve_all_features_with_filter( targeting_key, ) - # Convert from JSON strings to Python values - return { k: json.loads(v) for k, v in result.items() } + return self._decode_flags(result) except Exception as e: logger.error(f"Error resolving features: {e}") raise + @staticmethod + def _decode_flags(result: Dict[str, str]) -> Dict[str, Any]: + """Decode the JSON-encoded values the FFI cache hands back. + + Decoded per key so a single malformed value names the flag it came from, rather than + failing the whole evaluation with a bare JSONDecodeError. + """ + decoded: Dict[str, Any] = {} + for key, value in result.items(): + try: + decoded[key] = json.loads(value) + except json.JSONDecodeError as e: + raise SuperpositionError.serialization_error( + f"Flag '{key}' does not hold well-formed JSON: {value}", e + ) from e + return decoded + async def resolve_all_features_with_filter_async( self, context: Optional[EvaluationContext], @@ -200,7 +245,7 @@ async def resolve_all_features_with_filter_async( await self._ensure_fresh_data() if not self.ffi_cache or not self.cached_config: - raise RuntimeError("Provider not properly initialized") + raise SuperpositionError.provider_error("Provider not initialized: no cached config available") # Merge contexts targeting_key, query_data = self._merge_contexts(context) @@ -214,8 +259,7 @@ async def resolve_all_features_with_filter_async( targeting_key, ) - # Convert from JSON strings to Python values - return { k: json.loads(v) for k, v in result.items() } + return self._decode_flags(result) except Exception as e: logger.error(f"Error resolving features: {e}") raise @@ -238,8 +282,9 @@ async def get_applicable_variants( """ await self._ensure_fresh_data() - if not self.cached_experiments or not self.ffi_cache: - raise RuntimeError("Provider not properly initialized") + if not self.ffi_cache or not self.cached_experiments: + # No experiments cached means nothing can apply — not an error. + return [] # Merge contexts targeting_key, query_data = self._merge_contexts(context) @@ -357,7 +402,77 @@ async def refresh(self) -> None: """Manually refresh both config and experiments. Useful for MANUAL refresh strategy. + + Bounded by the refresh strategy's timeout, if it sets one: without it a data source that + never answers would stall the polling loop, or the caller under ON_DEMAND, indefinitely. + + Every refresh path — polling, watch, on-demand and manual — funnels through here, so this + is also where staleness is recorded. + + Raises: + SuperpositionError: if the refresh outlives the strategy's timeout. """ + succeeded = False + try: + timeout = self._refresh_timeout() + if timeout is None: + await self._refresh_once() + else: + try: + await asyncio.wait_for(self._refresh_once(), timeout=timeout / 1000) + except asyncio.TimeoutError as e: + logger.warning( + f"Refresh timed out after {timeout}ms, keeping last known good data" + ) + raise SuperpositionError.refresh_error( + f"Refresh timed out after {timeout}ms", e + ) from e + succeeded = True + finally: + self._record_refresh_outcome(succeeded) + + def _record_refresh_outcome(self, succeeded: bool) -> None: + """Mark the provider STALE while a failed refresh leaves the cache frozen. + + The flags keep resolving to their last known good values, and this is the only signal a + consumer has that they stopped tracking the source of truth. The next successful refresh + clears it. + + Only meaningful from READY: a failure during init is an ERROR (there is no good data to be + stale), and a provider that has been shut down stays NOT_READY. + + The event matters as much as the attribute: the SDK keeps its own copy of provider status + in `provider_registry._provider_status` and never reads ours, so without emitting, nothing + going through the OpenFeature client — `get_provider_status()`, an `on_provider_stale` + handler — would ever see this. Evaluation is unaffected either way: the client only + short-circuits on NOT_READY and FATAL. + """ + if succeeded: + if self.status == ProviderStatus.STALE: + logger.info("LocalResolutionProvider: refresh recovered, no longer stale") + self.status = ProviderStatus.READY + self.emit_provider_ready( + ProviderEventDetails(message="Refresh recovered; flags are current again") + ) + elif self.status == ProviderStatus.READY: + logger.warning("LocalResolutionProvider: refresh failed, serving stale data") + self.status = ProviderStatus.STALE + self.emit_provider_stale( + ProviderEventDetails( + message="Refresh failed; serving the last known good config" + ) + ) + + def _refresh_timeout(self) -> Optional[int]: + """The timeout the configured strategy puts on a single refresh, if any.""" + match self.refresh_strategy: + case PollingStrategy() | OnDemandStrategy(): + return self.refresh_strategy.timeout_ms() + case _: + return None + + async def _refresh_once(self) -> None: + """Fetch config and experiments concurrently.""" await asyncio.gather(self._fetch_and_cache_config(), self._fetch_and_cache_experiments()) # --- Private helpers --- @@ -381,20 +496,24 @@ async def _fetch_and_cache_config(self, init: bool = False) -> None: """Fetch and cache configuration from primary/fallback sources.""" try: logger.info(f"Fetching config (init={init})...") - # Try primary source - if_modified_since = None if init else self.cached_config.fetched_at + # Try primary source. A refresh after a failed init has no cached copy to date + # from, so it must ask for everything. + if_modified_since = ( + None if init or self.cached_config is None + else self.cached_config.fetched_at + ) response = await self.primary_source.fetch_config(if_modified_since) logger.debug(f"Primary source fetch_config response: {response}") + # The fetch returned, so the cache is confirmed current — whether it came back with + # new data or a 304. Either way the TTL clock restarts. + self.config_checked_at = datetime.now(timezone.utc) if response.get_data(): self.cached_config = response.get_data() self._update_config_ffi_cache() logger.info("Config updated from primary source") - elif init: - # need to counter the hack present in HttpDataSource - await self._handle_fetch_config_from_fallback() if not self.cached_config: - raise RuntimeError("Failed to fetch config from both primary and fallback sources") + raise SuperpositionError.config_error("Failed to fetch config from both primary and fallback sources") except Exception as e: logger.error(f"Error fetching config from primary source: {e}, init={init}") # Try fallback source if available @@ -432,20 +551,19 @@ async def _fetch_and_cache_experiments(self, init: bool = False) -> None: else self.cached_experiments.fetched_at ) response = await self.primary_source.fetch_active_experiments(if_modified_since) + # See _fetch_and_cache_config: new data or a 304, both are a successful check. + self.experiments_checked_at = datetime.now(timezone.utc) if response.get_data(): self.cached_experiments = response.get_data() self._update_exp_ffi_cache() logger.info("Experiments updated from primary source") - elif init: - # need to counter the hack present in HttpDataSource - await self._handle_fetch_experiments_from_fallback() if (not self.fallback_source or self.fallback_source.supports_experiments()) and not self.cached_experiments: - raise RuntimeError("Failed to fetch experiments from both primary and fallback sources") + raise SuperpositionError.config_error("Failed to fetch experiments from both primary and fallback sources") except Exception as e: logger.error(f"Error fetching experiments from primary source: {e}") - # Try fallback source if available and no 304 + # Try fallback source if available if init: await self._handle_fetch_experiments_from_fallback() else: @@ -458,17 +576,22 @@ async def _ensure_fresh_data(self) -> None: """Check if data needs refresh (for ON_DEMAND strategy).""" match self.refresh_strategy: case OnDemandStrategy(): - ttl = self.refresh_strategy.ttl + ttl = self.refresh_strategy.ttl_ms() use_stale_on_error = self.refresh_strategy.use_stale_on_error def is_elapsed(cached_at: datetime) -> bool: - return (datetime.now(timezone.utc) - cached_at).total_seconds() > ttl - - should_refresh_config = self.cached_config.fetched_at is None or is_elapsed(self.cached_config.fetched_at) - should_refresh_experiments = ( - self.cached_experiments is None - or self.cached_experiments.fetched_at is None - or is_elapsed(self.cached_experiments.fetched_at) + elapsed_ms = (datetime.now(timezone.utc) - cached_at).total_seconds() * 1000 + return elapsed_ms > ttl + + # Never checked, or last checked before the TTL window opened. Note this also + # removes an AttributeError: the old form read self.cached_config.fetched_at + # without guarding against cached_config being None after a failed init. + should_refresh_config = ( + self.config_checked_at is None or is_elapsed(self.config_checked_at) + ) + should_refresh_experiments = self.primary_source.supports_experiments() and ( + self.experiments_checked_at is None + or is_elapsed(self.experiments_checked_at) ) if should_refresh_config or should_refresh_experiments: @@ -491,20 +614,26 @@ async def _polling_loop() -> None: self_ref = weak_self() if self_ref is None: return - interval = self_ref.refresh_strategy.interval + interval = self_ref.refresh_strategy.interval_ms() del self_ref - logger.info(f"Starting polling with interval {interval}s") + logger.info(f"Starting polling with interval {interval}ms") try: while True: - await asyncio.sleep(interval) + await asyncio.sleep(interval / 1000) self_ref = weak_self() if self_ref is None: logger.info("LocalResolutionProvider has been garbage collected, stopping polling loop.") return - await self_ref.refresh() + try: + await self_ref.refresh() + except asyncio.CancelledError: + raise + except Exception as e: + # Keep polling on failure; the last known good data stays in place. + logger.warning(f"Polling refresh failed: {e}") del self_ref except asyncio.CancelledError: logger.info("Polling loop cancelled") @@ -556,7 +685,13 @@ async def _watch_loop() -> None: return logger.debug("Debounce settled, refreshing...") - await self_ref.refresh() + try: + await self_ref.refresh() + except asyncio.CancelledError: + raise + except Exception as e: + # Keep watching on failure; the last known good data stays in place. + logger.warning(f"Watch refresh failed: {e}") del self_ref except asyncio.CancelledError: logger.info("Watch loop cancelled") @@ -617,7 +752,7 @@ async def fetch_filtered_config( FetchResponse with ConfigData or NotModified status. """ if not self.ffi_cache or not self.cached_config: - raise RuntimeError("Provider not properly initialized or no config available") + raise SuperpositionError.data_source_error("No cached config available") if if_modified_since is not None: logger.debug("LocalResolutionProvider: ignoring if_modified_since, always reading fresh from file") @@ -640,10 +775,10 @@ async def fetch_active_experiments( FetchResponse with ExperimentData or NotModified status. """ if not self.supports_experiments(): - raise NotImplementedError("Experiments not supported by this provider") + raise SuperpositionError.data_source_error("Experiments not supported by this provider") if not self.cached_experiments: - raise RuntimeError("Provider not properly initialized or no experiments available") + raise SuperpositionError.data_source_error("No cached experiments available") if if_modified_since is not None: logger.debug("LocalResolutionProvider: ignoring if_modified_since for experiments, always returning cached data") @@ -658,10 +793,10 @@ async def fetch_candidate_active_experiments( ) -> FetchResponse[ExperimentData]: """Fetch candidate active experiments.""" if not self.supports_experiments(): - raise NotImplementedError("Experiments not supported by this provider") + raise SuperpositionError.data_source_error("Experiments not supported by this provider") if not self.ffi_cache or not self.cached_experiments: - raise RuntimeError("Provider not properly initialized or no experiments available") + raise SuperpositionError.data_source_error("No cached experiments available") if if_modified_since is not None: logger.debug("LocalResolutionProvider: ignoring if_modified_since for experiments, always returning cached data") @@ -679,10 +814,10 @@ async def fetch_matching_active_experiments( ) -> FetchResponse[ExperimentData]: """Fetch matching active experiments.""" if not self.supports_experiments(): - raise NotImplementedError("Experiments not supported by this provider") + raise SuperpositionError.data_source_error("Experiments not supported by this provider") if not self.ffi_cache or not self.cached_experiments: - raise RuntimeError("Provider not properly initialized or no experiments available") + raise SuperpositionError.data_source_error("No cached experiments available") if if_modified_since is not None: logger.debug("LocalResolutionProvider: ignoring if_modified_since for experiments, always returning cached data") diff --git a/clients/python/provider/superposition_provider/provider.py b/clients/python/provider/superposition_provider/provider.py index a1852f121..adf731e23 100644 --- a/clients/python/provider/superposition_provider/provider.py +++ b/clients/python/provider/superposition_provider/provider.py @@ -41,12 +41,9 @@ async def initialize(self, context: Optional[EvaluationContext] = None): cac_options=ConfigurationOptions( refresh_strategy=self.options.refresh_strategy, fallback_config=self.options.fallback_config, - evaluation_cache_options=self.options.evaluation_cache_options ), exp_options=ExperimentationOptions( refresh_strategy=self.options.experimentation_options.refresh_strategy, - evaluation_cache_options=self.options.experimentation_options.evaluation_cache_options, - default_toss=self.options.experimentation_options.default_toss ) ) await self.client.create_config() diff --git a/clients/python/provider/superposition_provider/remote_provider.py b/clients/python/provider/superposition_provider/remote_provider.py index 179a506ce..fc4b23bd0 100644 --- a/clients/python/provider/superposition_provider/remote_provider.py +++ b/clients/python/provider/superposition_provider/remote_provider.py @@ -76,7 +76,7 @@ def initialize(self, evaluation_context: EvaluationContext): logger.info("SuperpositionAPIProvider initialized successfully") except Exception as e: logger.error(f"Failed to initialize SuperpositionAPIProvider: {e}") - self.status = ProviderStatus.FATAL + self.status = ProviderStatus.ERROR raise def shutdown(self): @@ -146,16 +146,12 @@ async def get_applicable_variants( """ targeting_key, merged_context = self._get_merged_context(context) - if not targeting_key: - logger.warning("Missing targeting key for variant resolution") - return [] - try: response = await self.client.applicable_variants( input=ApplicableVariantsInput( workspace_id=self.options.workspace_id, org_id=self.options.org_id, - identifier=targeting_key, + identifier=targeting_key or "", context=merged_context, prefix=prefix_filter, ) diff --git a/clients/python/provider/superposition_provider/types.py b/clients/python/provider/superposition_provider/types.py index 6f89399dc..11ac7af15 100644 --- a/clients/python/provider/superposition_provider/types.py +++ b/clients/python/provider/superposition_provider/types.py @@ -2,9 +2,9 @@ Type definitions for Superposition provider configuration. """ -from dataclasses import dataclass, field +import warnings +from dataclasses import dataclass from typing import Optional, Dict, Any, Union -from .data_source import SuperpositionDataSource # ============================================================================ @@ -21,19 +21,40 @@ class SuperpositionOptions: # ============================================================================ -# Cache Configuration +# Refresh Strategy Types +# +# Durations are MILLISECONDS, matching the other Superposition clients. The seconds-based +# `interval` / `ttl` / `timeout` fields are DEPRECATED: set the `_milliseconds` field instead. +# Setting both units for the same duration raises ValueError — see `_reject_both`. # ============================================================================ -@dataclass -class EvaluationCacheOptions: - """Options for evaluation result caching.""" - ttl: Optional[int] = None - size: Optional[int] = None - +def _resolve_ms(milliseconds: Optional[int], seconds: Optional[int], name: str) -> Optional[int]: + """Prefer the millisecond field; fall back to the deprecated seconds one.""" + if milliseconds is not None: + return milliseconds + if seconds is None: + return None + warnings.warn( + f"'{name}' is deprecated and is read as SECONDS; use '{name}_milliseconds' instead.", + DeprecationWarning, + stacklevel=3, + ) + return seconds * 1000 + + +def _reject_both(milliseconds: Optional[int], seconds: Optional[int], name: str) -> None: + """Setting both units is ambiguous, so refuse it rather than silently picking one. + + Without this, starting from a default and overriding only the deprecated field — + `dataclasses.replace(default_polling_strategy(), interval=30)` — would leave the default's + `interval_milliseconds` in place, and it would win: the caller asks for 30s and gets 60s. + """ + if milliseconds is not None and seconds is not None: + raise ValueError( + f"set either '{name}' (deprecated, seconds) or '{name}_milliseconds', not both; " + f"'{name}_milliseconds' would win and '{name}' would be silently ignored" + ) -# ============================================================================ -# Refresh Strategy Types -# ============================================================================ @dataclass class PollingStrategy: @@ -41,11 +62,28 @@ class PollingStrategy: Fetches configuration at regular intervals. """ - interval: int # seconds - timeout: Optional[int] = None + interval: Optional[int] = None # DEPRECATED: seconds + timeout: Optional[int] = None # DEPRECATED: seconds + interval_milliseconds: Optional[int] = None + timeout_milliseconds: Optional[int] = None + + def __post_init__(self): + _reject_both(self.interval_milliseconds, self.interval, "interval") + _reject_both(self.timeout_milliseconds, self.timeout, "timeout") + + def interval_ms(self) -> int: + """The refresh interval in milliseconds.""" + resolved = _resolve_ms(self.interval_milliseconds, self.interval, "interval") + if resolved is None: + raise ValueError("PollingStrategy needs interval_milliseconds") + return resolved + + def timeout_ms(self) -> Optional[int]: + """The refresh timeout in milliseconds, if one is set.""" + return _resolve_ms(self.timeout_milliseconds, self.timeout, "timeout") def default_polling_strategy(): - return PollingStrategy(60, 30) + return PollingStrategy(interval_milliseconds=60_000, timeout_milliseconds=30_000) @dataclass class OnDemandStrategy: @@ -53,12 +91,31 @@ class OnDemandStrategy: Refreshes only when data becomes stale. """ - ttl: int # time-to-live in seconds - use_stale_on_error: bool = False - timeout: Optional[int] = None + ttl: Optional[int] = None # DEPRECATED: seconds + use_stale_on_error: bool = True + timeout: Optional[int] = None # DEPRECATED: seconds + ttl_milliseconds: Optional[int] = None + timeout_milliseconds: Optional[int] = None + + def __post_init__(self): + _reject_both(self.ttl_milliseconds, self.ttl, "ttl") + _reject_both(self.timeout_milliseconds, self.timeout, "timeout") + + def ttl_ms(self) -> int: + """How long cached data stays fresh, in milliseconds.""" + resolved = _resolve_ms(self.ttl_milliseconds, self.ttl, "ttl") + if resolved is None: + raise ValueError("OnDemandStrategy needs ttl_milliseconds") + return resolved + + def timeout_ms(self) -> Optional[int]: + """The refresh timeout in milliseconds, if one is set.""" + return _resolve_ms(self.timeout_milliseconds, self.timeout, "timeout") def default_on_demand_strategy(): - return OnDemandStrategy(300, True, 30) + return OnDemandStrategy( + use_stale_on_error=True, ttl_milliseconds=300_000, timeout_milliseconds=30_000 + ) @dataclass class WatchStrategy: @@ -92,34 +149,18 @@ class ManualStrategy: class ExperimentationOptions: """Configuration for experimentation client.""" refresh_strategy: RefreshStrategy - evaluation_cache_options: Optional[EvaluationCacheOptions] = None - default_toss: int = -1 @dataclass class ConfigurationOptions: """Configuration for config/CAC client.""" refresh_strategy: RefreshStrategy fallback_config: Optional[Dict[str, Any]] = None - evaluation_cache_options: Optional[EvaluationCacheOptions] = None + # ============================================================================ # Provider Initialization Options # ============================================================================ -@dataclass -class LocalProviderOptions: - """Options for LocalResolutionProvider.""" - primary_source: SuperpositionDataSource - fallback_source: Optional[SuperpositionDataSource] = None - refresh_strategy: RefreshStrategy = field(default_factory=default_polling_strategy) - - -@dataclass -class RemoteProviderOptions: - """Options for SuperpositionAPIProvider.""" - superposition_options: SuperpositionOptions - - @dataclass class SuperpositionProviderOptions: """Universal provider options (backward compatibility).""" @@ -130,5 +171,4 @@ class SuperpositionProviderOptions: workspace_id: str fallback_config: Optional[Dict[str, Any]] = None - evaluation_cache_options: Optional[EvaluationCacheOptions] = None experimentation_options: Optional[ExperimentationOptions] = None diff --git a/clients/python/provider/test_file_watch.py b/clients/python/provider/test_file_watch.py new file mode 100644 index 000000000..2038fb8a5 --- /dev/null +++ b/clients/python/provider/test_file_watch.py @@ -0,0 +1,116 @@ +"""FileDataSource supports several concurrent watchers, like the Java and Rust clients. + +One OS watcher, N subscriber queues. watch() used to build a new Observer per call and keep it in +a single field, so a second caller overwrote the first: when either generator exited, its `finally` +stopped whichever observer happened to be in the field, killing the other subscriber's watcher and +orphaning its own — a leaked OS handle and a subscriber that silently went deaf. + +Run standalone: python test_file_watch.py +""" + +import asyncio +import tempfile +from pathlib import Path + +from superposition_provider.file_data_source import FileDataSource + +TOML = """ +[default-configs] +currency = { value = "Rupee", schema = { type = "string" } } +[dimensions] +city = { position = 1, schema = { type = "string" }, type = "REGULAR" } +""" + + +async def _drive(generators, path: Path, body: str, timeout: float = 20.0): + """Advance each generator by one event, by making a change they will all see.""" + pending = [asyncio.ensure_future(g.__anext__()) for g in generators] + await asyncio.sleep(0.4) # let the watcher settle before touching the file + path.write_text(body) + return [await asyncio.wait_for(p, timeout=timeout) for p in pending] + + +def test_every_subscriber_sees_every_change(): + async def run(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.toml" + path.write_text(TOML) + source = FileDataSource(str(path)) + + first = source.watch() + second = source.watch() + + seen = await _drive([first, second], path, TOML + "\n# one\n") + assert all(str(path) in s for s in seen), "a subscriber missed the change" + + # One OS watcher shared by both, not one each. + assert len(source._subscribers) == 2 + assert source._observer is not None and source._observer.is_alive() + + await first.aclose() + await second.aclose() + await source.close() + + asyncio.run(run()) + + +def test_closing_one_subscriber_leaves_the_other_working(): + async def run(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.toml" + path.write_text(TOML) + source = FileDataSource(str(path)) + + first = source.watch() + second = source.watch() + await _drive([first, second], path, TOML + "\n# one\n") + + observer = source._observer + await first.aclose() + + # The shared watcher must survive while a subscriber remains. + assert len(source._subscribers) == 1, "closing one subscriber dropped the other" + assert source._observer is observer and observer.is_alive(), ( + "closing one subscriber stopped the shared watcher" + ) + + # ...and the survivor still gets events. + seen = await _drive([second], path, TOML + "\n# two\n") + assert str(path) in seen[0] + + await second.aclose() + await source.close() + + asyncio.run(run()) + + +def test_the_watcher_stops_once_the_last_subscriber_leaves(): + async def run(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.toml" + path.write_text(TOML) + source = FileDataSource(str(path)) + + first = source.watch() + second = source.watch() + await _drive([first, second], path, TOML + "\n# one\n") + observer = source._observer + + await first.aclose() + assert observer.is_alive() + + await second.aclose() + assert source._observer is None, "the watcher outlived its last subscriber" + assert not observer.is_alive(), "the OS handle was leaked" + + await source.close() + + asyncio.run(run()) + + +if __name__ == "__main__": + cases = [v for k, v in sorted(globals().items()) if k.startswith("test_")] + for case in cases: + case() + print(f"ok {case.__name__}") + print(f"\n{len(cases)} passed") diff --git a/clients/python/provider/test_http_data_source.py b/clients/python/provider/test_http_data_source.py new file mode 100644 index 000000000..1dc70e9f0 --- /dev/null +++ b/clients/python/provider/test_http_data_source.py @@ -0,0 +1,116 @@ +"""HttpDataSource distinguishes "nothing changed" from "the call failed". + +The SDK's deserializer treats every status >= 300 as an error and collapses it into an +UnknownApiError carrying only a message. This used to be caught wholesale and reported as +FetchResponse.not_modified(), so expired credentials and 500s looked like a successful 304 +and silently froze the config. These tests pin the difference. + +Run standalone: python test_http_data_source.py +""" + +import asyncio +import json +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer + +from superposition_provider.http_data_source import HttpDataSource +from superposition_provider.types import SuperpositionOptions + +# Set per-test; the handler replies with whatever is here. +STATUS = 200 +BODY = {} + + +def _reply(self): + # A 304 carries no body, by definition. + if STATUS == 304: + self.send_response(304) + self.end_headers() + return + payload = json.dumps(BODY).encode() + self.send_response(STATUS) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + +class _Handler(BaseHTTPRequestHandler): + # The SDK posts to /config; answer whatever verb it uses. + do_GET = do_POST = do_PUT = do_DELETE = _reply + + def log_message(self, *args): + pass # keep the test output readable + + +def _serve(): + server = HTTPServer(("127.0.0.1", 0), _Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + return server + + +def _fetch_config(server): + # The SDK builds an aiohttp session eagerly, so the source must be created on the loop. + async def run(): + source = HttpDataSource(SuperpositionOptions( + endpoint=f"http://127.0.0.1:{server.server_address[1]}", + token="test-token", + org_id="test-org", + workspace_id="test-workspace", + )) + return await source.fetch_config(None) + + return asyncio.run(run()) + + +def test_a_304_is_not_modified(): + global STATUS + server = _serve() + try: + STATUS = 304 + response = _fetch_config(server) + assert response.is_not_modified(), "304 should be NotModified" + assert response.get_data() is None + finally: + server.shutdown() + + +def test_an_auth_failure_is_not_reported_as_not_modified(): + global STATUS, BODY + server = _serve() + try: + STATUS, BODY = 401, {"message": "Unauthorized"} + try: + response = _fetch_config(server) + except Exception: + return # the only acceptable outcome + raise AssertionError( + f"401 surfaced as {response!r} instead of raising — a caller would treat " + "expired credentials as 'config unchanged' and serve stale flags forever" + ) + finally: + STATUS, BODY = 200, {} + server.shutdown() + + +def test_a_server_error_is_not_reported_as_not_modified(): + global STATUS, BODY + server = _serve() + try: + STATUS, BODY = 500, {"message": "boom"} + try: + response = _fetch_config(server) + except Exception: + return + raise AssertionError(f"500 surfaced as {response!r} instead of raising") + finally: + STATUS, BODY = 200, {} + server.shutdown() + + +if __name__ == "__main__": + cases = [v for k, v in sorted(globals().items()) if k.startswith("test_")] + for case in cases: + case() + print(f"ok {case.__name__}") + print(f"\n{len(cases)} passed") diff --git a/clients/python/provider/test_ondemand_ttl.py b/clients/python/provider/test_ondemand_ttl.py new file mode 100644 index 000000000..6acf9cba5 --- /dev/null +++ b/clients/python/provider/test_ondemand_ttl.py @@ -0,0 +1,135 @@ +"""The ON_DEMAND TTL is measured from the last *check*, not the config's last *change*. + +An HTTP data source stamps ConfigData.fetched_at with the server's last-modified — when the config +last changed. Driving the TTL off that made a config which had been stable for longer than the TTL +permanently "stale": every evaluation fired a fetch, the 304 that came back left the timestamp +untouched, and the next evaluation fired another. The more stable the config, the more load. + +FileDataSource stamps datetime.now() instead, which is why the file-based paths never showed this. +ServerLike below reproduces the HTTP behaviour. + +Run standalone: python test_ondemand_ttl.py +""" + +import asyncio +import time +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, List, Optional + +from openfeature.evaluation_context import EvaluationContext + +from superposition_provider.data_source import ( + ConfigData, + FetchResponse, + SuperpositionDataSource, +) +from superposition_provider.local_provider import LocalResolutionProvider +from superposition_provider.types import OnDemandStrategy + +from superposition_bindings.superposition_types import Config + +# The config last changed an hour ago — and is perfectly current. +SERVER_LAST_MODIFIED = datetime.now(timezone.utc) - timedelta(hours=1) + + +class ServerLike(SuperpositionDataSource): + """Behaves like HttpDataSource: fetched_at is the server's last-modified, repeats get a 304.""" + + def __init__(self): + self.fetches = 0 + + async def fetch_filtered_config( + self, + context: Optional[Dict[str, Any]] = None, + prefix_filter: Optional[List[str]] = None, + if_modified_since: Optional[datetime] = None, + ) -> FetchResponse[ConfigData]: + self.fetches += 1 + if if_modified_since is not None: + return FetchResponse.not_modified() + config = Config( + contexts=[], + overrides={}, + default_configs={"currency": '"Rupee"'}, + dimensions={}, + ) + return FetchResponse.data(ConfigData(data=config, fetched_at=SERVER_LAST_MODIFIED)) + + async def fetch_active_experiments(self, if_modified_since=None): + raise NotImplementedError + + async def fetch_candidate_active_experiments(self, context=None, prefix_filter=None, if_modified_since=None): + raise NotImplementedError + + async def fetch_matching_active_experiments(self, context=None, prefix_filter=None, if_modified_since=None): + raise NotImplementedError + + def supports_experiments(self) -> bool: + return False + + async def close(self) -> None: + pass + + +def _provider(ttl_ms: int): + source = ServerLike() + provider = LocalResolutionProvider( + source, + refresh_strategy=OnDemandStrategy(ttl_milliseconds=ttl_ms, timeout_milliseconds=5_000), + ) + return source, provider + + +def test_an_unchanged_config_is_not_refetched_within_the_ttl(): + async def run(): + source, provider = _provider(60_000) + await provider.initialize(EvaluationContext()) + after_init = source.fetches + + for _ in range(5): + await provider.resolve_all_features_async(EvaluationContext()) + + assert source.fetches == after_init, ( + f"evaluations inside the TTL hit the source {source.fetches - after_init} times; " + "however old the config is, they should hit it zero times" + ) + await provider.shutdown() + + asyncio.run(run()) + + +def test_the_ttl_still_expires_and_a_304_restarts_it(): + async def run(): + source, provider = _provider(100) + await provider.initialize(EvaluationContext()) + after_init = source.fetches + + # Inside the TTL: free. + await provider.resolve_all_features_async(EvaluationContext()) + assert source.fetches == after_init + + time.sleep(0.2) + + # Past the TTL: exactly one re-check. + await provider.resolve_all_features_async(EvaluationContext()) + assert source.fetches == after_init + 1, "an expired TTL must re-check the source" + + # The 304 it got back is a successful check, so the clock restarts and this is free. + await provider.resolve_all_features_async(EvaluationContext()) + assert source.fetches == after_init + 1, "a 304 is a successful check and must restart the TTL" + + # And the cache kept serving the last known good value throughout. + flags = await provider.resolve_all_features_async(EvaluationContext()) + assert flags["currency"] == "Rupee" + + await provider.shutdown() + + asyncio.run(run()) + + +if __name__ == "__main__": + cases = [v for k, v in sorted(globals().items()) if k.startswith("test_")] + for case in cases: + case() + print(f"ok {case.__name__}") + print(f"\n{len(cases)} passed") diff --git a/clients/python/provider/test_type_contract.py b/clients/python/provider/test_type_contract.py new file mode 100644 index 000000000..8daceb02a --- /dev/null +++ b/clients/python/provider/test_type_contract.py @@ -0,0 +1,103 @@ +"""The flag type contract, shared with the Rust and Java clients. + +No coercion. An integer widens to a float; nothing else converts. Every case below is a +value some client used to read as the wrong type — Python coerced "true" and 1 to booleans, +Java truncated 1.5 to an integer, and Rust turned a top-level array into an index-keyed map. + +These exercise the extractors directly, so no server or FFI cache is needed. +""" + +import json +from typing import Any, Dict, List, Optional + +from openfeature.evaluation_context import EvaluationContext +from openfeature.exception import ErrorCode + +from superposition_provider.interfaces import AllFeatureProvider + +FLAGS = { + "currency": "Rupee", + "price": 10000, + "ratio": 1.5, + "enabled": True, + "tags": ["a", "b"], + "meta": {"tier": "gold", "credits": 5}, +} + + +class StubProvider(AllFeatureProvider): + """Serves a fixed flag set, mirroring what the FFI cache hands back.""" + + def resolve_all_features_with_filter( + self, + context: Optional[EvaluationContext] = None, + prefix_filter: Optional[List[str]] = None, + ) -> Dict[str, Any]: + # The real provider json.loads the FFI's JSON-encoded strings; round-trip to match. + return {k: json.loads(json.dumps(v)) for k, v in FLAGS.items()} + + async def resolve_all_features_with_filter_async( + self, + context: Optional[EvaluationContext] = None, + prefix_filter: Optional[List[str]] = None, + ) -> Dict[str, Any]: + return self.resolve_all_features_with_filter(context, prefix_filter) + + +provider = StubProvider() +ctx = EvaluationContext() + + +def test_values_resolve_as_their_own_type(): + assert provider.resolve_string("currency", "none", ctx).value == "Rupee" + assert provider.resolve_int("price", 0, ctx).value == 10000 + assert provider.resolve_float("ratio", 0.0, ctx).value == 1.5 + assert provider.resolve_bool("enabled", False, ctx).value is True + assert provider.resolve_object("meta", {}, ctx).value == {"tier": "gold", "credits": 5} + assert provider.resolve_object("tags", [], ctx).value == ["a", "b"] + + +def test_an_integer_widens_to_a_float(): + # Lossless, and the one conversion all three clients allow. + assert provider.resolve_float("price", 0.0, ctx).value == 10000.0 + + +def test_a_float_does_not_narrow_to_an_integer(): + # Java used to truncate this to 1. + result = provider.resolve_int("ratio", 0, ctx) + assert result.error_code == ErrorCode.TYPE_MISMATCH + assert result.value == 0 + + +def test_strings_and_numbers_are_not_booleans(): + # _to_bool used to accept "true" and any non-zero number. + assert provider.resolve_bool("currency", False, ctx).error_code == ErrorCode.TYPE_MISMATCH + assert provider.resolve_bool("price", False, ctx).error_code == ErrorCode.TYPE_MISMATCH + + +def test_a_primitive_is_not_an_object(): + assert provider.resolve_object("currency", {}, ctx).error_code == ErrorCode.TYPE_MISMATCH + + +def test_a_number_is_not_a_string(): + assert provider.resolve_string("price", "none", ctx).error_code == ErrorCode.TYPE_MISMATCH + + +def test_a_missing_flag_is_flag_not_found_not_a_type_mismatch(): + result = provider.resolve_string("nope", "fallback", ctx) + assert result.error_code == ErrorCode.FLAG_NOT_FOUND + assert result.value == "fallback" + + +def test_the_default_is_returned_on_every_error(): + assert provider.resolve_bool("currency", True, ctx).value is True + assert provider.resolve_int("ratio", 42, ctx).value == 42 + + +if __name__ == "__main__": + # The package has no pytest dependency, so these run standalone too. + cases = [v for k, v in sorted(globals().items()) if k.startswith("test_")] + for case in cases: + case() + print(f"ok {case.__name__}") + print(f"\n{len(cases)} passed")