diff --git a/README.md b/README.md index ab4923ba..f5d307a3 100644 --- a/README.md +++ b/README.md @@ -239,6 +239,46 @@ result = client.organizations.list_organizations( > [!NOTE] > The WorkOS API currently honors `Idempotency-Key` only on the [Create Audit Log Event](https://workos.com/docs/reference/audit-logs/event) endpoint (`audit_logs.create_event`). Other endpoints accept the header but do not deduplicate requests, so a retried mutation elsewhere can still create a duplicate. +## Feature Flags Runtime Client + +The runtime client keeps an environment's feature flag configuration in memory and evaluates flags locally, with no network request per evaluation. It polls the configuration in the background (every 30 seconds by default) with the client's API key, base URL and retry settings. Create one long-lived instance per process and share it across requests. It is for trusted server-side code only. + +```python +from workos import WorkOSClient + +workos = WorkOSClient(api_key="sk_...") +flags = workos.create_feature_flags_runtime_client() +flags.start() # polls on a daemon thread +flags.wait_until_ready(timeout=5) # raises TimeoutError if no configuration arrives + +flags.is_enabled("new-checkout", {"user": "user_123", "organization": "org_123"}) +flags.is_enabled("new-checkout", {"workspace": "ws_1"}, default=True) +flags.get_all_flags({"organization": "org_123"}) # {"new-checkout": True, ...} + +flags.close() # on shutdown; or use `with workos.create_feature_flags_runtime_client() as flags:` +``` + +**Evaluation.** The context maps a target type to the exact, case-sensitive ID being evaluated. `user` and `organization` are target types like any [custom target](https://workos.com/docs/feature-flags/custom-targets) type, and an ID of `None` is skipped. A disabled flag serves its off value. For an enabled flag, the first rule whose conditions all match wins, so a rule serving `False` can exclude a user before a later rule includes their organization; otherwise the flag's default value applies. A missing flag, an unready client, or a flag without a boolean answer returns `default` (`False` unless you pass one), and `get_all_flags` omits flags without an answer. Evaluation never raises: invalid context entries are ignored with a logged warning and the rest is evaluated. + +**Failures.** A failed poll keeps the last good configuration and retries with exponential backoff. Transient failures (429, 500, 502, 503, 504, timeouts, connection errors) are also retried within one refresh, up to the client's `max_retries` and within `request_timeout`. A 429's `Retry-After` is honored up to 60 seconds, both by those retries and before the next poll. The client streams the response and gives every network wait only the time left, so a refresh takes at most `request_timeout` plus one read of a response body that stalls; a custom `HTTPBackend` that is not an httpx client receives the remaining time as its per-request timeout instead. A 401, from a poll or from `refresh()`, stops polling for good: create a new client with a valid API key. Failures are logged to the `workos.feature_flags.runtime` logger (pass `logger=` to use your own) and reported to error listeners. `flags.status` shows readiness, the last error, refresh counts and timestamps, `polling_stopped`, `flag_count` and `cache_age()`. + +**Notifications.** `on_change` (or `add_change_listener`) fires after a refresh installs a different configuration, including the first load, with the previous and current snapshots and the changed slugs. `on_error` fires for failed background polls and `on_status` after every refresh. Listeners run one at a time; their exceptions are logged and never affect polling. + +**Bootstrap.** Persist the configuration and pass it back to make a new client ready before its first poll: + +```python +from workos import FeatureFlagsConfiguration + +flags = workos.create_feature_flags_runtime_client( + bootstrap=FeatureFlagsConfiguration.from_json(cached_json), # raises on unreadable JSON + on_change=lambda event: save_cache(event.current.to_json()), +) +``` + +**Inspection and manual refresh.** `flags.get_flag(slug)` returns a flag's configuration as received, with its rules in order. `flags.refresh()` fetches the configuration immediately, raising on failure, and works without `start()`, e.g. in scripts or serverless functions. + +**Processes.** The runtime client polls through the `WorkOSClient`'s HTTP client, so close it before closing the `WorkOSClient` or an HTTP client you supplied. Create the `WorkOSClient` and the runtime client in each worker process, after forking, rather than in an app gunicorn loads with `--preload`. Background threads must be enabled, so run uWSGI with `--enable-threads`. + ## Type Safety This SDK ships with full type annotations (`py.typed` / PEP 561) and works with mypy, pyright, and IDE autocompletion out of the box. All API resource models are `@dataclass(slots=True)` classes with `from_dict()` / `to_dict()` for serialization. diff --git a/src/workos/__init__.py b/src/workos/__init__.py index 1a1182f4..f29ac8fe 100644 --- a/src/workos/__init__.py +++ b/src/workos/__init__.py @@ -23,6 +23,10 @@ TransportTimeout, ) from ._pagination import AsyncPage, ListMetadata, SyncPage +from .feature_flags.runtime import ( + FeatureFlagsConfiguration, + FeatureFlagsRuntimeClient, +) from .public_client import create_public_client from ._types import NOT_GIVEN, NotGiven, RequestOptions @@ -51,4 +55,6 @@ "TransportError", "TransportTimeout", "TransportConnectError", + "FeatureFlagsRuntimeClient", + "FeatureFlagsConfiguration", ] diff --git a/src/workos/_client.py b/src/workos/_client.py index 9776dd23..592c6480 100644 --- a/src/workos/_client.py +++ b/src/workos/_client.py @@ -190,6 +190,12 @@ def pkce(self) -> PKCE: """PKCE (Proof Key for Code Exchange) utilities.""" return PKCE() + # A plain function bound as a method, so its signature and docs live next + # to the runtime client's constructor. + from .feature_flags.runtime._factory import ( + create_feature_flags_runtime_client as create_feature_flags_runtime_client, + ) + # @oagen-ignore-end diff --git a/src/workos/feature_flags/runtime/__init__.py b/src/workos/feature_flags/runtime/__init__.py index bcc4452a..1b488922 100644 --- a/src/workos/feature_flags/runtime/__init__.py +++ b/src/workos/feature_flags/runtime/__init__.py @@ -2,4 +2,30 @@ # This file is hand-maintained. The feature flags runtime client evaluates # flags locally from a polled configuration; it is not part of the OpenAPI spec. -"""Local evaluation of WorkOS feature flags from a polled configuration.""" +"""Local evaluation of WorkOS feature flags from a polled configuration. + +Use :class:`FeatureFlagsRuntimeClient` with :class:`workos.WorkOSClient`. +""" + +from ._configuration import ( + ConditionDefinition, + FeatureFlagsConfiguration, + FeatureFlagsConfigurationError, + FlagDefinition, + RuleDefinition, +) +from ._context import EvaluationContext +from ._sync import FeatureFlagsRuntimeClient +from ._types import FlagsChangedEvent, RuntimeStatus + +__all__ = [ + "ConditionDefinition", + "EvaluationContext", + "FeatureFlagsConfiguration", + "FeatureFlagsConfigurationError", + "FeatureFlagsRuntimeClient", + "FlagDefinition", + "FlagsChangedEvent", + "RuleDefinition", + "RuntimeStatus", +] diff --git a/src/workos/feature_flags/runtime/_core.py b/src/workos/feature_flags/runtime/_core.py new file mode 100644 index 00000000..cfd9fb6d --- /dev/null +++ b/src/workos/feature_flags/runtime/_core.py @@ -0,0 +1,332 @@ +# @oagen-ignore-file +# This file is hand-maintained. State and local evaluation for the feature +# flags runtime client. +# pyright: strict, reportPrivateUsage=false + +from __future__ import annotations + +import copy +import logging +import threading +from collections import deque +from dataclasses import replace +from datetime import datetime, timezone +from typing import TYPE_CHECKING, Any, Callable, Deque, Optional, Tuple + +from ..._errors import APIError +from . import _policy +from ._configuration import ( + FeatureFlagsConfiguration, + FeatureFlagsConfigurationError, + FlagDefinition, + changed_flags, + configuration_for, +) +from ._context import ContextWarnings, EvaluationContext, LoggerLike, normalize_context +from ._types import FlagsChangedEvent, RuntimeStatus + +if TYPE_CHECKING: + from ..._base_client import _BaseWorkOSClient + +LOGGER_NAME = "workos.feature_flags.runtime" + +Listener = Callable[[Any], object] +Notification = Tuple[Listener, Any] + + +def utc_now() -> datetime: + return datetime.now(timezone.utc) + + +class RuntimeClientCore: + """Readiness, status, listeners and local evaluation for a runtime client. + + The configuration snapshot is immutable and replaced by a single reference + assignment, so evaluation never locks and always sees one complete + snapshot, even while a refresh replaces it on another thread. + """ + + def __init__( + self, + client: _BaseWorkOSClient, + *, + polling_interval: float, + request_timeout: float, + max_retry_delay: float, + bootstrap: Optional[FeatureFlagsConfiguration], + logger: Optional[LoggerLike], + ) -> None: + # A client built for public (PKCE) flows has no API key, and every + # poll would fail with a 401. Fail before starting instead. + client._require_api_key() + _require_configuration(bootstrap) + self._polling_interval = _policy.polling_interval(polling_interval) + self._request_timeout = _policy.validate_duration( + request_timeout, "request_timeout" + ) + self._max_retry_delay = _policy.validate_duration( + max_retry_delay, "max_retry_delay" + ) + # Retries of transient failures inside one refresh follow the client's + # `max_retries`, as the Node SDK's transport retries do. + self._max_retries = client._resolve_max_retries(None) + self._logger: LoggerLike = ( + logger if logger is not None else logging.getLogger(LOGGER_NAME) + ) + self._context_warnings = ContextWarnings(self._logger) + self._status = RuntimeStatus(configuration=bootstrap) + self._status_lock = threading.Lock() + self._listeners_lock = threading.Lock() + self._listeners: dict[str, tuple[Listener, ...]] = { + "change": (), + "error": (), + "status": (), + } + self._started = False + self._closed = False + # Notification batches, appended in refresh order while the refresh + # lock is held and delivered after it is released, so a listener may + # call refresh() itself. + self._pending: Deque[list[Notification]] = deque() + # The 401 that stopped polling before any configuration was available. + self._readiness_error: Optional[Exception] = None + + # --- Evaluation ----------------------------------------------------------- + + def is_enabled( + self, + slug: str, + context: Optional[EvaluationContext] = None, + *, + default: bool = False, + ) -> bool: + """Evaluate one flag locally, without a network request. + + A disabled flag serves its off value. For an enabled flag, the first rule + whose conditions all match the context wins; otherwise the flag's + default value applies. ``default`` is returned for a missing flag, an + unready client, and a flag without a boolean answer. Never raises: + invalid context entries are ignored with a logged warning. + + Args: + slug: The flag's slug. + context: Target type to exact ID, e.g. ``{"user": "user_123", + "organization": "org_123", "workspace": "ws_1"}``. + default: Returned when the flag has no answer. + """ + configuration = self._status.configuration + flag = configuration._compiled(slug) if configuration is not None else None + if flag is None: + return default + answer = flag.evaluate(normalize_context(context, self._context_warnings)) + return default if answer is None else answer + + def get_all_flags( + self, context: Optional[EvaluationContext] = None + ) -> dict[str, bool]: + """Evaluate every flag in one snapshot for the given context. + + Flags without a boolean answer, such as a future multivariate flag, are + omitted so that ``flags.get(slug, my_default)`` applies your default. + An unready client returns an empty dict. + """ + configuration = self._status.configuration + if configuration is None: + return {} + targets = normalize_context(context, self._context_warnings) + evaluated: dict[str, bool] = {} + for slug, flag in configuration._compiled_items(): + answer = flag.evaluate(targets) + if answer is not None: + evaluated[slug] = answer + return evaluated + + def get_flag(self, slug: str) -> Optional[FlagDefinition]: + """Return one flag's configuration from the current snapshot, for inspection.""" + configuration = self._status.configuration + return configuration.get_flag(slug) if configuration is not None else None + + @property + def status(self) -> RuntimeStatus: + """Readiness, the current snapshot, refresh timestamps, counts and the last error.""" + return self._status + + @property + def configuration(self) -> Optional[FeatureFlagsConfiguration]: + """The current snapshot, suitable for persisting with ``to_json()``.""" + return self._status.configuration + + @property + def is_ready(self) -> bool: + """Whether a validated configuration is available for local evaluation.""" + return self._status.configuration is not None + + @property + def is_closed(self) -> bool: + """Whether ``close()`` has been called.""" + return self._closed + + # --- Listeners ------------------------------------------------------------ + + def _add_listener(self, kind: str, listener: Listener) -> None: + if not callable(listener): + raise TypeError(f"{kind} listener must be callable") + with self._listeners_lock: + self._listeners[kind] = (*self._listeners[kind], listener) + + def _remove_listener(self, kind: str, listener: Listener) -> None: + with self._listeners_lock: + self._listeners[kind] = tuple( + registered + for registered in self._listeners[kind] + if registered != listener + ) + + def _notifications(self, kind: str, argument: object) -> list[Notification]: + return [(listener, argument) for listener in self._listeners[kind]] + + # --- Refresh outcomes ----------------------------------------------------- + + def _record_success( + self, + configuration: FeatureFlagsConfiguration, + attempted: datetime, + *, + background: bool, + ) -> list[Notification]: + with self._status_lock: + previous = self._status + current = replace( + previous, + configuration=configuration, + last_successful_refresh_at=utc_now(), + last_refresh_attempt_at=attempted, + last_error=None, + refresh_count=previous.refresh_count + 1, + ) + self._status = current + if background and previous.last_error is not None: + self._logger.info("Feature flag refresh recovered") + self._logger.debug( + "Feature flag refresh succeeded (%d flags)", len(configuration) + ) + notifications = self._notifications("status", current) + # An unchanged payload keeps the same snapshot. A new one can differ + # only outside its flags, and a cache persisted from change events + # still needs it. + if configuration is not previous.configuration: + changed = changed_flags(previous.configuration, configuration) + event = FlagsChangedEvent(previous.configuration, configuration, changed) + notifications += self._notifications("change", event) + return notifications + + def _record_failure( + self, error: Exception, attempted: datetime, *, background: bool + ) -> list[Notification]: + with self._status_lock: + previous = self._status + current = replace( + previous, + last_refresh_attempt_at=attempted, + last_error=error, + refresh_count=previous.refresh_count + 1, + failed_refresh_count=previous.failed_refresh_count + 1, + ) + self._status = current + notifications = self._notifications("status", current) + if not background: + self._logger.debug("Feature flag refresh failed. %s", describe_error(error)) + return notifications + if not _policy.is_unauthorized(error): + # A 401 is logged once polling stops; see _record_polling_stopped. + level = logging.ERROR if _status_code(error) == 403 else logging.WARNING + self._logger.log( + level, + "Feature flag refresh failed; keeping the last good flags and retrying. %s", + describe_error(error), + ) + return notifications + self._notifications("error", error) + + def _record_polling_stopped(self, error: Exception) -> list[Notification]: + with self._status_lock: + current = replace(self._status, polling_stopped=True) + self._status = current + self._logger.error( + "Feature flag polling stopped after a 401; keeping the last good flags. " + "Create a new runtime client with a valid API key to resume. %s", + describe_error(error), + ) + return self._notifications("status", current) + + def _parse(self, payload: object) -> FeatureFlagsConfiguration: + # The transport returns None for a body that is not JSON; the + # configuration rejects that like any other payload it can't read. + try: + return configuration_for(payload, self._status.configuration) + except FeatureFlagsConfigurationError as error: + raise FeatureFlagsConfigurationError( + "Feature flag poll returned a payload this SDK can't read; " + f"keeping the last good flags. {error}" + ) from error + + def _readiness_timeout(self, timeout: Optional[float]) -> TimeoutError: + last_error = self._status.last_error + message = f"Feature flags were not ready within {timeout} seconds" + if last_error is not None: + message += f"; the last refresh failed: {describe_error(last_error)}" + error = TimeoutError(message) + error.__cause__ = last_error + error.__suppress_context__ = True + return error + + def _readiness_failure(self) -> Exception: + """A new exception for each waiter, so no shared traceback keeps growing.""" + stored = self._readiness_error + if stored is None: + return RuntimeError( + "The feature flags runtime client was closed before it became ready" + ) + try: + # Same type and attributes, with an empty traceback. + error = copy.copy(stored) + except Exception: + error = RuntimeError( + f"Feature flag polling stopped before it was ready: {describe_error(stored)}" + ) + error.__cause__ = stored + error.__suppress_context__ = True + return error + + def _log_listener_error(self) -> None: + self._logger.exception("A feature flags runtime client listener raised") + + +def describe_error(error: BaseException) -> str: + """A one-line description for logs, with the HTTP status of API errors.""" + if isinstance(error, APIError): + parts = [type(error).__name__] + if error.status_code is not None: + parts.append(f"HTTP {error.status_code}") + if error.message and error.message != "No message": + parts.append(error.message) + description = ": ".join(parts) + return ( + f"{description} (request {error.request_id})" + if error.request_id + else description + ) + message = str(error) + return f"{type(error).__name__}: {message}" if message else type(error).__name__ + + +def _status_code(error: Exception) -> Optional[int]: + status_code = getattr(error, "status_code", None) + return status_code if isinstance(status_code, int) else None + + +def _require_configuration(bootstrap: object) -> None: + if bootstrap is not None and not isinstance(bootstrap, FeatureFlagsConfiguration): + raise TypeError( + "bootstrap must be a FeatureFlagsConfiguration; parse cached JSON with " + "FeatureFlagsConfiguration.from_json()" + ) diff --git a/src/workos/feature_flags/runtime/_factory.py b/src/workos/feature_flags/runtime/_factory.py new file mode 100644 index 00000000..f9966a80 --- /dev/null +++ b/src/workos/feature_flags/runtime/_factory.py @@ -0,0 +1,49 @@ +# @oagen-ignore-file +# This file is hand-maintained. The factory method that WorkOSClient exposes +# as `create_feature_flags_runtime_client`. +# pyright: strict + +from __future__ import annotations + +from typing import TYPE_CHECKING, Callable, Optional + +from . import _policy +from ._configuration import FeatureFlagsConfiguration +from ._context import LoggerLike +from ._sync import FeatureFlagsRuntimeClient +from ._types import FlagsChangedEvent, RuntimeStatus + +if TYPE_CHECKING: + from ..._base_client import WorkOSClient + + +def create_feature_flags_runtime_client( + self: WorkOSClient, + *, + polling_interval: float = _policy.DEFAULT_POLLING_INTERVAL, + request_timeout: float = _policy.DEFAULT_REQUEST_TIMEOUT, + max_retry_delay: float = _policy.DEFAULT_MAX_RETRY_DELAY, + bootstrap: Optional[FeatureFlagsConfiguration] = None, + logger: Optional[LoggerLike] = None, + on_change: Optional[Callable[[FlagsChangedEvent], None]] = None, + on_error: Optional[Callable[[Exception], None]] = None, + on_status: Optional[Callable[[RuntimeStatus], None]] = None, +) -> FeatureFlagsRuntimeClient: + """Create a feature flags runtime client that polls through this client. + + The runtime client evaluates flags locally from a configuration it polls in + the background. Call ``start()`` to begin polling. See + :class:`~workos.feature_flags.runtime.FeatureFlagsRuntimeClient` for the + options. + """ + return FeatureFlagsRuntimeClient( + self, + polling_interval=polling_interval, + request_timeout=request_timeout, + max_retry_delay=max_retry_delay, + bootstrap=bootstrap, + logger=logger, + on_change=on_change, + on_error=on_error, + on_status=on_status, + ) diff --git a/src/workos/feature_flags/runtime/_policy.py b/src/workos/feature_flags/runtime/_policy.py new file mode 100644 index 00000000..395473bf --- /dev/null +++ b/src/workos/feature_flags/runtime/_policy.py @@ -0,0 +1,121 @@ +# @oagen-ignore-file +# This file is hand-maintained. It holds the polling and retry policy of the +# feature flags runtime client. +# pyright: strict + +from __future__ import annotations + +import math +import random + +from ..._base_client import RETRY_STATUS_CODES +from ..._errors import ( + APIError, + RateLimitExceededError, + WorkOSConnectionError, + WorkOSError, + WorkOSTimeoutError, +) +from ..._http import TransportError + +# Matches the Node and .NET runtime clients. +DEFAULT_POLLING_INTERVAL = 30.0 +MINIMUM_POLLING_INTERVAL = 5.0 +DEFAULT_REQUEST_TIMEOUT = 10.0 +DEFAULT_MAX_RETRY_DELAY = 60.0 +MINIMUM_DELAY = 1.0 +INITIAL_RETRY_DELAY = 1.0 +JITTER_FACTOR = 0.1 +MAX_COUNTED_FAILURES = 20 + +# Retries inside one refresh: at most the client's `max_retries`, like the +# Node SDK's transport retries, and always within the refresh's request timeout. +_ATTEMPT_RETRY_BASE = 0.5 +# A retry is only worth sending with at least this much of the budget left +# for the request itself. +MINIMUM_ATTEMPT_TIME = 0.5 +# Matches the Node SDK's MAXIMUM_RETRY_AFTER_TIME_IN_MILLISECONDS. +MAXIMUM_RETRY_AFTER = 60.0 +_ATTEMPT_RETRY_MULTIPLIER = 1.5 + +POLL_PATH = ("sdk", "feature-flags") +POLL_PARAMS = {"payload_version": "2"} + + +def validate_duration(value: object, name: str) -> float: + """Reject durations a timer cannot honor.""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise TypeError(f"{name} must be a number of seconds") + if not math.isfinite(value) or value <= 0: + raise ValueError(f"{name} must be a positive, finite number of seconds") + return float(value) + + +def polling_interval(value: float) -> float: + return max(MINIMUM_POLLING_INTERVAL, validate_duration(value, "polling_interval")) + + +def next_delay(interval: float, failures: int, max_retry_delay: float) -> float: + """Seconds until the next poll. + + Failures back off exponentially from one second up to ``max_retry_delay``, + but never poll sooner than the polling interval. A ten percent jitter in + either direction spreads polls from many processes, with a one second floor. + """ + base = interval + if failures > 0: + exponent = min(failures, MAX_COUNTED_FAILURES) - 1 + base = max(base, min(INITIAL_RETRY_DELAY * 2**exponent, max_retry_delay)) + jitter = 1 + (random.random() * 2 - 1) * JITTER_FACTOR + return max(MINIMUM_DELAY, base * jitter) + + +def is_unauthorized(error: BaseException) -> bool: + """A 401 stops polling: the API key is invalid and retrying cannot help.""" + return isinstance(error, APIError) and error.status_code == 401 + + +def attempt_retry_delay( + error: BaseException, attempt: int, remaining: float, max_retries: int +) -> float | None: + """Seconds to wait before retrying a failed request inside one refresh. + + ``None`` means give up and fail the refresh with ``error``: it is not + transient, ``max_retries`` retries were already made, or the wait would not + leave ``MINIMUM_ATTEMPT_TIME`` of the budget for the retry itself. A retry + sent with almost no time left would only time out and hide the server's + real error. + """ + if attempt >= max_retries or not _is_transient(error): + return None + requested = retry_after(error) + if requested is not None: + delay = requested + else: + delay = ( + _ATTEMPT_RETRY_BASE + * _ATTEMPT_RETRY_MULTIPLIER**attempt + * (0.5 + random.random()) + ) + return delay if delay + MINIMUM_ATTEMPT_TIME <= remaining else None + + +def retry_after(error: BaseException) -> float | None: + """The wait a 429 asked for, capped at ``MAXIMUM_RETRY_AFTER`` seconds. + + A retry inside a refresh waits this long. A refresh that could not honor it + within its budget fails; the next poll then waits at least this long. + """ + if isinstance(error, RateLimitExceededError) and error.retry_after is not None: + return min(max(error.retry_after, 0.0), MAXIMUM_RETRY_AFTER) + return None + + +def _is_transient(error: BaseException) -> bool: + if isinstance(error, APIError): + return error.status_code in RETRY_STATUS_CODES + if isinstance(error, (WorkOSTimeoutError, WorkOSConnectionError)): + return True + return isinstance(error, WorkOSError) and isinstance( + error.__cause__, TransportError + ) diff --git a/src/workos/feature_flags/runtime/_sync.py b/src/workos/feature_flags/runtime/_sync.py new file mode 100644 index 00000000..1dcc8c7c --- /dev/null +++ b/src/workos/feature_flags/runtime/_sync.py @@ -0,0 +1,406 @@ +# @oagen-ignore-file +# This file is hand-maintained. The synchronous feature flags runtime client +# polls on a background thread and evaluates flags locally. +# pyright: strict, reportPrivateUsage=false + +from __future__ import annotations + +import inspect +import threading +import time +from types import TracebackType +from typing import TYPE_CHECKING, Any, Callable, Optional, Type + +from . import _policy +from ._configuration import FeatureFlagsConfiguration +from ._context import LoggerLike +from ._core import Notification, RuntimeClientCore, utc_now +from ._transport import PollTransport +from ._types import FlagsChangedEvent, RuntimeStatus + +if TYPE_CHECKING: + from ..._base_client import WorkOSClient + + +class _Closed(Exception): + """The client was closed while a refresh was in progress.""" + + +class FeatureFlagsRuntimeClient(RuntimeClientCore): + """Polls an environment's feature flag configuration and evaluates flags locally. + + Create one long-lived instance per process and share it across requests. + Evaluation never makes a network request and never raises. Polling runs on + a daemon thread started by :meth:`start` and stopped by :meth:`close`:: + + flags = workos.create_feature_flags_runtime_client() + flags.start() + flags.wait_until_ready(timeout=5) + flags.is_enabled("new-checkout", {"organization": "org_123"}) + + Polling uses the ``WorkOSClient``'s API key, base URL, headers, retry + settings and HTTP client, so close this client before closing the + ``WorkOSClient`` or an HTTP client you supplied. + """ + + def __init__( + self, + client: WorkOSClient, + *, + polling_interval: float = _policy.DEFAULT_POLLING_INTERVAL, + request_timeout: float = _policy.DEFAULT_REQUEST_TIMEOUT, + max_retry_delay: float = _policy.DEFAULT_MAX_RETRY_DELAY, + bootstrap: Optional[FeatureFlagsConfiguration] = None, + logger: Optional[LoggerLike] = None, + on_change: Optional[Callable[[FlagsChangedEvent], None]] = None, + on_error: Optional[Callable[[Exception], None]] = None, + on_status: Optional[Callable[[RuntimeStatus], None]] = None, + ) -> None: + """Create a runtime client. No request is made until :meth:`start` or :meth:`refresh`. + + Args: + client: The ``WorkOSClient`` whose API key, base URL and HTTP client to poll with. + polling_interval: Seconds between polls, before a 10% jitter. At least 5. + request_timeout: Seconds one refresh may take, including retries of + transient failures. The response is streamed and every network + wait gets only the time left, so a refresh ends within this, plus + at most one read of a response body that stalls. A custom + ``HTTPBackend`` that is not an httpx client gets the remaining + time as its per-request timeout instead. + max_retry_delay: Upper bound in seconds for the exponential backoff + after failed polls. Polls never come sooner than the interval. + bootstrap: A cached configuration to evaluate from before the first + successful poll. Makes the client ready immediately. + logger: Where to log failures and ignored context. Defaults to the + ``workos.feature_flags.runtime`` logger. + on_change: Called after a refresh installs a different configuration, + including the first load. + on_error: Called when a background poll fails. + on_status: Called after every refresh, successful or not, and when + polling stops. + + Raises: + ConfigurationError: If ``client`` has no API key. + """ + super().__init__( + client, + polling_interval=polling_interval, + request_timeout=request_timeout, + max_retry_delay=max_retry_delay, + bootstrap=bootstrap, + logger=logger, + ) + self._client = client + self._transport = PollTransport(client) + self._thread: Optional[threading.Thread] = None + self._delivering: Optional[int] = None + self._init_synchronization() + if on_change is not None: + self.add_change_listener(on_change) + if on_error is not None: + self.add_error_listener(on_error) + if on_status is not None: + self.add_status_listener(on_status) + + def _init_synchronization(self) -> None: + self._lifecycle_lock = threading.Lock() + self._refresh_lock = threading.RLock() + self._drain_lock = threading.Lock() + # `_stop` is set by close(); `_wake` also interrupts the poller's sleep + # when polling stops after a 401. + self._stop = threading.Event() + self._wake = threading.Event() + self._ready = threading.Event() + if self._status.ready: + self._ready.set() + + # --- Lifecycle ------------------------------------------------------------ + + def start(self) -> None: + """Start background polling. Returns immediately and is idempotent. + + The first poll is sent right away, then every ``polling_interval`` + seconds. Failed polls retry with exponential backoff. A 401 stops + polling permanently; the last good configuration stays usable. + + Raises: + RuntimeError: If the client is closed. + """ + with self._lifecycle_lock: + if self._closed: + raise RuntimeError( + "This feature flags runtime client is closed; create a new one" + ) + if self._started: + return + self._started = True + if self._status.polling_stopped: + self._logger.warning( + "Not starting feature flag polling: it stopped after a 401. " + "Create a new runtime client with a valid API key." + ) + return + self._spawn_poller() + + def wait_until_ready(self, timeout: Optional[float] = None) -> None: + """Block until a bootstrap or a successful refresh makes flags available. + + Does not start polling: call :meth:`start` (or :meth:`refresh`), possibly + from another thread. + + Args: + timeout: Seconds to wait. ``None`` waits indefinitely. + + Raises: + TimeoutError: If flags are not available within ``timeout``. + AuthenticationError: If polling stopped after a 401 before any + configuration was available. + RuntimeError: If the client was closed before it became ready. + """ + if self._status.ready: + return + if not self._ready.wait(timeout): + raise self._readiness_timeout(timeout) + if not self._status.ready: + raise self._readiness_failure() + + def refresh(self) -> None: + """Fetch the configuration now, on the calling thread. + + Serialized with background polling. Errors are raised to the caller and + the last good configuration is kept; ``on_error`` is not called. A 401 + also stops background polling, as it would from a background poll. + + Raises: + WorkOSError: If the request fails or the payload cannot be read. + RuntimeError: If the client is closed. + """ + if self._closed: + raise RuntimeError( + "This feature flags runtime client is closed; create a new one" + ) + try: + self._refresh(background=False) + except _Closed: + raise RuntimeError( + "The feature flags runtime client was closed during the refresh" + ) from None + except Exception as error: + if _policy.is_unauthorized(error): + self._stop_polling(error) + raise + + def close(self) -> None: + """Stop polling and release the background thread. Idempotent. + + Waits up to ``request_timeout`` for an in-flight poll to finish. A poll + still running after that, such as one reading a stalled response, is + abandoned: it ends on its own and changes nothing. Every caller waits, + including one that calls while another close() is still waiting. + Evaluation keeps serving the last configuration after closing. + """ + with self._lifecycle_lock: + if not self._closed: + self._closed = True + self._stop.set() + self._wake.set() + self._ready.set() + thread = self._thread + if ( + thread is not None + and thread is not threading.current_thread() + and self._delivering != threading.get_ident() + and thread.is_alive() + ): + thread.join(self._request_timeout + 1.0) + + def __enter__(self) -> FeatureFlagsRuntimeClient: + self.start() + return self + + def __exit__( + self, + exc_type: Optional[Type[BaseException]], + exc_val: Optional[BaseException], + exc_tb: Optional[TracebackType], + ) -> None: + self.close() + + # --- Listeners ------------------------------------------------------------ + + def add_change_listener( + self, listener: Callable[[FlagsChangedEvent], None] + ) -> None: + """Call ``listener`` after a refresh installs a different configuration. + + Fires on the first load and on the first refresh that differs from a + bootstrap; never for identical refreshes or failures. Persist + ``event.current.to_json()`` here to keep a bootstrap cache fresh. + Listeners run one at a time in refresh order, on the thread that + refreshed or on one already delivering notifications; their exceptions + are logged and never affect polling. + """ + self._add_listener("change", listener) + + def remove_change_listener( + self, listener: Callable[[FlagsChangedEvent], None] + ) -> None: + """Stop calling a change listener.""" + self._remove_listener("change", listener) + + def add_error_listener(self, listener: Callable[[Exception], None]) -> None: + """Call ``listener`` with the error of every failed background poll.""" + self._add_listener("error", listener) + + def remove_error_listener(self, listener: Callable[[Exception], None]) -> None: + """Stop calling an error listener.""" + self._remove_listener("error", listener) + + def add_status_listener(self, listener: Callable[[RuntimeStatus], None]) -> None: + """Call ``listener`` with the new status after every refresh and when polling stops.""" + self._add_listener("status", listener) + + def remove_status_listener(self, listener: Callable[[RuntimeStatus], None]) -> None: + """Stop calling a status listener.""" + self._remove_listener("status", listener) + + # --- Polling -------------------------------------------------------------- + + def _spawn_poller(self) -> None: + thread = threading.Thread( + target=self._poll_forever, name="workos-feature-flags", daemon=True + ) + self._thread = thread + thread.start() + + def _poll_forever(self) -> None: + failures = 0 + while not self._stop.is_set() and not self._status.polling_stopped: + retry_after: Optional[float] = None + try: + self._refresh(background=True) + failures = 0 + except _Closed: + return + except Exception as error: + if self._stop.is_set(): + return + if _policy.is_unauthorized(error): + self._stop_polling(error) + return + failures = min(failures + 1, _policy.MAX_COUNTED_FAILURES) + retry_after = _policy.retry_after(error) + delay = _policy.next_delay( + self._polling_interval, failures, self._max_retry_delay + ) + if self._wake.wait(max(delay, retry_after or 0.0)): + return + + def _refresh(self, *, background: bool) -> None: + try: + with self._refresh_lock: + self._refresh_locked(background=background) + finally: + self._drain() + + def _refresh_locked(self, *, background: bool) -> None: + if self._closed or (background and self._status.polling_stopped): + raise _Closed() + attempted = utc_now() + # close() sets `_closed` under the lifecycle lock, so checking it and + # recording the outcome under that lock means nothing a refresh does + # lands after close() returns. + try: + configuration = self._parse(self._fetch()) + except _Closed: + raise + except Exception as error: + with self._lifecycle_lock: + if self._closed: + raise _Closed() from error + self._pending.append( + self._record_failure(error, attempted, background=background) + ) + raise + with self._lifecycle_lock: + if self._closed: + raise _Closed() + notifications = self._record_success( + configuration, attempted, background=background + ) + self._ready.set() + self._pending.append(notifications) + + def _fetch(self) -> Any: + deadline = time.monotonic() + self._request_timeout + attempt = 0 + while True: + try: + return self._transport.get(deadline) + except Exception as error: + delay = _policy.attempt_retry_delay( + error, attempt, deadline - time.monotonic(), self._max_retries + ) + if delay is None: + raise + if self._stop.wait(delay): + raise _Closed() from error + attempt += 1 + + def _stop_polling(self, error: Exception) -> None: + with self._refresh_lock: + if self._status.polling_stopped: + return + notifications = self._record_polling_stopped(error) + if not self._status.ready: + self._readiness_error = error + self._wake.set() + self._ready.set() + self._pending.append(notifications) + self._drain() + + def _drain(self) -> None: + """Deliver pending notifications in refresh order, one batch at a time. + + Only one thread delivers at once. A refresh that finishes while + another thread is delivering leaves its batch to that thread, which + also covers a listener calling refresh(): its batch is delivered after + the current one, never interleaved with it. + """ + with self._drain_lock: + if self._delivering is not None: + return + self._delivering = threading.get_ident() + try: + while True: + with self._drain_lock: + if not self._pending: + self._delivering = None + return + batch = self._pending.popleft() + if not self._closed: + self._deliver(batch) + except BaseException: + with self._drain_lock: + self._delivering = None + raise + + def _deliver(self, notifications: list[Notification]) -> None: + for listener, argument in notifications: + try: + result = listener(argument) + except Exception: + self._log_listener_error() + continue + if inspect.isawaitable(result): + _discard_awaitable(result) + self._logger.warning( + "An async listener was registered on FeatureFlagsRuntimeClient and " + "was not run; listeners must be plain functions" + ) + + +def _discard_awaitable(awaitable: object) -> None: + close = getattr(awaitable, "close", None) + if callable(close): + close() diff --git a/src/workos/feature_flags/runtime/_transport.py b/src/workos/feature_flags/runtime/_transport.py new file mode 100644 index 00000000..9e123c22 --- /dev/null +++ b/src/workos/feature_flags/runtime/_transport.py @@ -0,0 +1,141 @@ +# @oagen-ignore-file +# This file is hand-maintained. The sync runtime client's poll request, sent +# with a total time budget. +# pyright: strict, reportPrivateUsage=false + +from __future__ import annotations + +import time +from typing import TYPE_CHECKING, Any, Dict, Optional +from urllib.parse import urlsplit + +from ..._errors import WorkOSConnectionError, WorkOSError, WorkOSTimeoutError +from ..._http import ( + HTTPResponse, + HttpxBackend, + TransportConnectError, + TransportError, + TransportTimeout, + _HttpxFamily, +) +from . import _policy + +if TYPE_CHECKING: + from ..._base_client import WorkOSClient + +_PHASES = ("connect", "read", "write", "pool") +_MINIMUM_WAIT = 0.001 + + +class _DeadlineTimeouts(Dict[str, Optional[float]]): + """httpx timeouts that shrink to the time left before a deadline. + + httpcore reads each phase's timeout when the phase starts, so connecting, + sending and waiting for the response each get only what is left of the + budget instead of the whole budget every time. + """ + + def __init__(self, deadline: float) -> None: + super().__init__({phase: None for phase in _PHASES}) + self._deadline = deadline + + def remaining(self) -> float: + return max(self._deadline - time.monotonic(), _MINIMUM_WAIT) + + def get(self, key: str, default: Optional[float] = None) -> Optional[float]: # type: ignore[override] + return self.remaining() if key in _PHASES else super().get(key, default) + + def __getitem__(self, key: str) -> Optional[float]: + return self.remaining() if key in _PHASES else super().__getitem__(key) + + +class PollTransport: + """Sends ``GET /sdk/feature-flags`` for the sync runtime client. + + The request carries the same URL, headers and User-Agent as any other + request from the ``WorkOSClient``, goes through its HTTP client, and maps + responses and failures to the same errors. Unlike other requests, a refresh + has a total budget. The body is streamed, every network wait is limited to + the time left, and the response is abandoned once the deadline passes, so a + refresh ends within ``request_timeout`` plus at most one read of a stalling + response body. A caller-supplied ``HTTPBackend`` that is not an httpx client + cannot stream; it gets the remaining budget as its per-request timeout. + """ + + def __init__(self, workos: WorkOSClient) -> None: + self._workos = workos + backend = workos._backend + self._shared_httpx: Any = None + self._family = _HttpxFamily("httpx2") + if isinstance(backend, HttpxBackend): + self._shared_httpx = backend._client + self._family = backend._family + + def get(self, deadline: float) -> Any: + """Send one poll request and return the decoded body.""" + workos = self._workos + url = workos._build_request_url(_policy.POLL_PATH, _policy.POLL_PARAMS, None) + headers = workos._build_headers("get", None, None) + try: + response = self._send(url, headers, deadline) + except TransportTimeout as error: + raise WorkOSTimeoutError(f"Request timed out: {error}") from error + except TransportConnectError as error: + raise WorkOSConnectionError(f"Connection failed: {error}") from error + except TransportError as error: + raise WorkOSError(f"Network error: {error}") from error + if response.status_code >= 400: + workos._raise_error(response) + return workos._deserialize_response(response, None) + + def _send(self, url: str, headers: Dict[str, str], deadline: float) -> HTTPResponse: + if deadline - time.monotonic() <= 0: + raise TransportTimeout("no time was left in the request timeout") + client = self._shared_httpx + if client is None: + return self._workos._backend.request( + "GET", + url, + headers=headers, + content=None, + timeout=max(deadline - time.monotonic(), _MINIMUM_WAIT), + ) + try: + return self._stream(client, url, headers, deadline) + except self._family.timeout_error as error: + raise TransportTimeout(str(error)) from error + except self._family.connect_error as error: + raise TransportConnectError(str(error)) from error + except self._family.http_error as error: + raise TransportError(str(error)) from error + + def _stream( + self, client: Any, url: str, headers: Dict[str, str], deadline: float + ) -> HTTPResponse: + request = client.build_request( + "GET", + url, + # Explicit params merge client defaults, as HttpxBackend does. + params=urlsplit(url).query or None, + headers=headers, + extensions={"timeout": _DeadlineTimeouts(deadline)}, + ) + response = client.send(request, stream=True) + try: + chunks: list[bytes] = [] + for chunk in response.iter_bytes(): + chunks.append(chunk) + if time.monotonic() > deadline: + raise TransportTimeout( + "the response body did not finish within the request timeout" + ) + content = b"".join(chunks) + finally: + response.close() + return HTTPResponse( + status_code=response.status_code, + headers=response.headers, + content=content, + request_method="GET", + request_url=str(response.request.url), + ) diff --git a/src/workos/feature_flags/runtime/_types.py b/src/workos/feature_flags/runtime/_types.py new file mode 100644 index 00000000..4b068a9d --- /dev/null +++ b/src/workos/feature_flags/runtime/_types.py @@ -0,0 +1,73 @@ +# @oagen-ignore-file +# This file is hand-maintained. Public value types of the feature flags +# runtime client. +# pyright: strict + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone + +from ._configuration import FeatureFlagsConfiguration + + +@dataclass(frozen=True, slots=True) +class FlagsChangedEvent: + """A refresh installed a configuration that differs from the previous one.""" + + previous: FeatureFlagsConfiguration | None + """The snapshot this refresh replaced. + + ``None`` on a first load without bootstrap, and the bootstrap configuration + on the first refresh after one. + """ + current: FeatureFlagsConfiguration + """The newly installed snapshot. Persist ``current.to_json()`` to bootstrap later.""" + changed_flags: frozenset[str] + """Slugs that were added, removed or modified. Every slug on a first load. + + Empty when only fields outside the flags changed. + """ + + +@dataclass(frozen=True, slots=True) +class RuntimeStatus: + """An immutable view of readiness and the most recent refresh result.""" + + configuration: FeatureFlagsConfiguration | None = None + """The last valid snapshot. Failed refreshes never replace it.""" + last_successful_refresh_at: datetime | None = None + """When the last successful network refresh finished. ``None`` for bootstrap-only readiness.""" + last_refresh_attempt_at: datetime | None = None + """When the most recent finished refresh started.""" + last_error: Exception | None = None + """The most recent refresh failure, cleared by the next successful refresh.""" + refresh_count: int = 0 + """Refreshes that finished, successfully or not, whether polled or manual.""" + failed_refresh_count: int = 0 + """Finished refreshes that failed, including timeouts and unreadable payloads.""" + polling_stopped: bool = False + """Background polling stopped after a 401. + + The last good snapshot stays usable but is never refreshed again. Fix the + API key and create a new runtime client. Closing the client does not set + this. + """ + + @property + def ready(self) -> bool: + """Whether a validated configuration is available for local evaluation.""" + return self.configuration is not None + + @property + def flag_count(self) -> int: + """Number of flags in the current snapshot.""" + return len(self.configuration) if self.configuration is not None else 0 + + def cache_age(self) -> float | None: + """Seconds since the last successful network refresh, or ``None`` if there was none.""" + if self.last_successful_refresh_at is None: + return None + return ( + datetime.now(timezone.utc) - self.last_successful_refresh_at + ).total_seconds() diff --git a/tests/feature_flags_runtime/conftest.py b/tests/feature_flags_runtime/conftest.py new file mode 100644 index 00000000..76e5f8d6 --- /dev/null +++ b/tests/feature_flags_runtime/conftest.py @@ -0,0 +1,67 @@ +# @oagen-ignore-file +"""Fixtures for the feature flags runtime client tests.""" + +from __future__ import annotations + +from typing import Any, Callable, Iterator, List + +import pytest + +from workos import WorkOSClient +from workos.feature_flags.runtime import FeatureFlagsRuntimeClient, _policy + +from tests.feature_flags_runtime.helpers import ( + API_KEY, + BASE_URL, + FlagServer, + SyncBackend, +) + + +@pytest.fixture +def server() -> FlagServer: + return FlagServer() + + +@pytest.fixture +def client(server: FlagServer) -> Iterator[WorkOSClient]: + workos = WorkOSClient( + api_key=API_KEY, base_url=BASE_URL, http_client=SyncBackend(server) + ) + yield workos + workos.close() + + +@pytest.fixture +def fast_polling(monkeypatch: pytest.MonkeyPatch) -> List[int]: + """Poll every 10 ms and retry inside a refresh after 5 ms. + + Returns the failure counts the poller computed delays for, so tests can + assert on backoff without waiting for it. + """ + failures: List[int] = [] + + def next_delay(interval: float, count: int, max_retry_delay: float) -> float: + failures.append(count) + return 0.01 + + monkeypatch.setattr(_policy, "next_delay", next_delay) + monkeypatch.setattr(_policy, "_ATTEMPT_RETRY_BASE", 0.005) + return failures + + +@pytest.fixture +def make_runtime( + client: WorkOSClient, +) -> Iterator[Callable[..., FeatureFlagsRuntimeClient]]: + """Create runtime clients that are always closed at teardown.""" + created: List[FeatureFlagsRuntimeClient] = [] + + def make(**options: Any) -> FeatureFlagsRuntimeClient: + runtime = client.create_feature_flags_runtime_client(**options) + created.append(runtime) + return runtime + + yield make + for runtime in created: + runtime.close() diff --git a/tests/feature_flags_runtime/helpers.py b/tests/feature_flags_runtime/helpers.py index 58e7ec78..8019e5aa 100644 --- a/tests/feature_flags_runtime/helpers.py +++ b/tests/feature_flags_runtime/helpers.py @@ -4,10 +4,19 @@ from __future__ import annotations import json +import threading +import time +from dataclasses import dataclass from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Callable, Dict, List, Optional, Union + +import httpx2 + +from workos import HTTPResponse FIXTURES = Path(__file__).parent / "fixtures" +API_KEY = "sk_test_runtime" +BASE_URL = "https://api.workos.test" def load_fixture(name: str) -> Dict[str, Any]: @@ -46,3 +55,217 @@ def rule( {"target_type": target_type, "operator": "one_of", "values": list(ids)} ], } + + +@dataclass +class RecordedRequest: + method: str + url: str + headers: Dict[str, str] + timeout: float + + +Reply = Union[tuple, Exception, Callable[[], Any]] + + +class FlagServer: + """Scripted stand-in for the API behind an HTTPBackend. + + Replies queued with ``enqueue`` are used once, in order; otherwise every + request gets the ``respond`` reply. A reply is ``(status, body)``, an + exception to raise, or ``BLOCK`` to hold the request until ``release()``. + """ + + BLOCK = object() + + def __init__(self) -> None: + self._condition = threading.Condition() + self._queue: List[Any] = [] + self._default: Any = (200, payload()) + self._released = threading.Event() + self.requests: List[RecordedRequest] = [] + self.in_flight = 0 + self.max_in_flight = 0 + + def respond(self, status: int, body: Any = None) -> None: + with self._condition: + self._default = (status, body) + + def enqueue(self, *replies: Any) -> None: + with self._condition: + self._queue.extend(replies) + + def release(self) -> None: + self._released.set() + + def wait_for_requests(self, count: int, timeout: float = 5.0) -> None: + with self._condition: + if not self._condition.wait_for( + lambda: len(self.requests) >= count, timeout + ): + raise AssertionError( + f"expected {count} requests, saw {len(self.requests)}" + ) + + def _next( + self, method: str, url: str, headers: Dict[str, str], timeout: float + ) -> Any: + with self._condition: + self.requests.append(RecordedRequest(method, url, dict(headers), timeout)) + self.in_flight += 1 + self.max_in_flight = max(self.max_in_flight, self.in_flight) + reply = self._queue.pop(0) if self._queue else self._default + self._condition.notify_all() + return reply + + def _finish(self) -> None: + with self._condition: + self.in_flight -= 1 + + def _response(self, reply: Any, method: str, url: str) -> HTTPResponse: + if isinstance(reply, Exception): + raise reply + status, body, *extra = reply + headers = { + "x-request-id": f"req_{len(self.requests)}", + **(extra[0] if extra else {}), + } + content = ( + body + if isinstance(body, bytes) + else b"" + if body is None + else json.dumps(body).encode() + ) + return HTTPResponse( + status_code=status, + headers=httpx2.Headers(headers), + content=content, + request_method=method, + request_url=url, + ) + + +class SyncBackend: + def __init__(self, server: FlagServer) -> None: + self.server = server + + def request( + self, + method: str, + url: str, + *, + headers: Dict[str, str], + content: Optional[bytes], + timeout: float, + ) -> HTTPResponse: + reply = self.server._next(method, url, headers, timeout) + try: + if reply is FlagServer.BLOCK: + self.server._released.wait(10) + reply = (200, payload()) + return self.server._response(reply, method, url) + finally: + self.server._finish() + + def close(self) -> None: + pass + + +class Collector: + """Thread-safe list of listener calls with a blocking wait.""" + + def __init__(self) -> None: + self._condition = threading.Condition() + self.items: List[Any] = [] + + def __call__(self, item: Any) -> None: + with self._condition: + self.items.append(item) + self._condition.notify_all() + + def wait_for(self, count: int, timeout: float = 5.0) -> List[Any]: + with self._condition: + if not self._condition.wait_for(lambda: len(self.items) >= count, timeout): + raise AssertionError( + f"expected {count} calls, saw {len(self.items)}: {self.items}" + ) + return list(self.items) + + +def eventually(predicate: Callable[[], bool], timeout: float = 5.0) -> None: + deadline = time.monotonic() + timeout + while not predicate(): + if time.monotonic() > deadline: + raise AssertionError("condition was not met in time") + time.sleep(0.002) + + +class LocalHTTPServer: + """A real HTTP/1.1 server on 127.0.0.1 for transport-level tests. + + ``mode`` selects the next responses: ``"ok"`` serves ``payload``, + ``"stall"`` waits ``stall`` seconds before answering, and ``"drip"`` sends + the body one byte every ``drip_interval`` seconds. Every request is + recorded with its headers. + """ + + def __init__(self, payload: Dict[str, Any]) -> None: + import http.server + + self.payload = payload + self.mode = "ok" + self.stall = 0.0 + self.drip_interval = 0.1 + self.requests: List[Dict[str, Any]] = [] + self._lock = threading.Lock() + owner = self + + class Handler(http.server.BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, format: str, *args: Any) -> None: + pass + + def do_GET(self) -> None: + owner._handle(self) + + self._server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self._server.daemon_threads = True + self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) + self._thread.start() + + @property + def base_url(self) -> str: + return f"http://127.0.0.1:{self._server.server_address[1]}" + + def close(self) -> None: + self._server.shutdown() + self._server.server_close() + + def _handle(self, handler: Any) -> None: + with self._lock: + self.requests.append( + { + "path": handler.path, + "headers": dict(handler.headers.items()), + } + ) + mode, stall, interval = self.mode, self.stall, self.drip_interval + if mode == "stall": + time.sleep(stall) + body = json.dumps(self.payload).encode() + handler.send_response(200) + handler.send_header("Content-Type", "application/json") + handler.send_header("Content-Length", str(len(body))) + handler.end_headers() + try: + if mode == "drip": + for byte in body: + handler.wfile.write(bytes([byte])) + handler.wfile.flush() + time.sleep(interval) + else: + handler.wfile.write(body) + except OSError: + pass diff --git a/tests/feature_flags_runtime/test_configuration.py b/tests/feature_flags_runtime/test_configuration.py index c5cc0eac..6e291d6c 100644 --- a/tests/feature_flags_runtime/test_configuration.py +++ b/tests/feature_flags_runtime/test_configuration.py @@ -7,16 +7,20 @@ import json import pickle from collections import UserDict +from datetime import datetime, timezone from types import MappingProxyType import pytest -from workos.feature_flags.runtime._configuration import ( +from workos._errors import AuthenticationError +from workos.feature_flags.runtime import ( ConditionDefinition, FeatureFlagsConfiguration, FeatureFlagsConfigurationError, FlagDefinition, + FlagsChangedEvent, RuleDefinition, + RuntimeStatus, ) from tests.feature_flags_runtime.helpers import flag, payload, rule @@ -148,3 +152,29 @@ def test_snapshots_pickle_and_copy() -> None: # Immutable, so copies can share the instance. assert copy.copy(configuration) is configuration assert copy.deepcopy(configuration) is configuration + + +def test_status_and_change_events_pickle_and_copy() -> None: + configuration = FeatureFlagsConfiguration.from_dict(PAYLOAD) + error = AuthenticationError( + "Invalid API key", request_id="req_1", response_json={"message": "x"} + ) + status = RuntimeStatus( + configuration=configuration, + last_successful_refresh_at=datetime(2026, 11, 2, tzinfo=timezone.utc), + last_error=error, + refresh_count=3, + failed_refresh_count=1, + ) + event = FlagsChangedEvent(None, configuration, frozenset({"checkout"})) + + for clone in (pickle.loads(pickle.dumps(status)), copy.deepcopy(status)): + assert clone.refresh_count == 3 + assert clone.flag_count == 1 + assert clone.configuration is not None + assert clone.configuration.to_json() == configuration.to_json() + assert isinstance(clone.last_error, AuthenticationError) + assert clone.last_error.request_id == "req_1" + for clone_event in (pickle.loads(pickle.dumps(event)), copy.deepcopy(event)): + assert clone_event.changed_flags == {"checkout"} + assert clone_event.current.slugs == ("checkout",) diff --git a/tests/feature_flags_runtime/test_corpus.py b/tests/feature_flags_runtime/test_corpus.py index db5e3df0..bb592d2d 100644 --- a/tests/feature_flags_runtime/test_corpus.py +++ b/tests/feature_flags_runtime/test_corpus.py @@ -4,18 +4,21 @@ from __future__ import annotations import logging -from typing import Any, Dict +from contextlib import contextmanager +from typing import Any, Dict, Iterator, List import pytest -from workos.feature_flags.runtime._configuration import ( +from workos import WorkOSClient +from workos.feature_flags.runtime import ( FeatureFlagsConfiguration, FeatureFlagsConfigurationError, - changed_flags, + FeatureFlagsRuntimeClient, ) +from workos.feature_flags.runtime._configuration import changed_flags from workos.feature_flags.runtime._context import ContextWarnings, normalize_context -from tests.feature_flags_runtime.helpers import load_fixture +from tests.feature_flags_runtime.helpers import API_KEY, load_fixture EVALUATION = load_fixture("evaluation_cases.json") PAYLOADS = load_fixture("payload_cases.json") @@ -26,29 +29,26 @@ _EVALUATION_FIELDS = {"context", "expect"} _OPTIONAL_EVALUATION_FIELDS = {"name"} _CHANGE_FIELDS = {"name", "source", "before", "after", "changed"} -_WARNINGS = ContextWarnings(logging.getLogger("tests.feature_flags_runtime.corpus")) -def _is_enabled( +@contextmanager +def _runtime( configuration: FeatureFlagsConfiguration, - slug: str, - context: Dict[str, Any], - *, - default: bool, -) -> bool: - # `default` answers for a missing flag and for one without a boolean answer. - flag = configuration._compiled(slug) - if flag is None: - return default - answer = flag.evaluate(normalize_context(context, _WARNINGS)) - return default if answer is None else answer +) -> Iterator[FeatureFlagsRuntimeClient]: + workos = WorkOSClient(api_key=API_KEY) + runtime = FeatureFlagsRuntimeClient(workos, bootstrap=configuration) + try: + yield runtime + finally: + runtime.close() + workos.close() def _observed( - configuration: FeatureFlagsConfiguration, slug: str, context: Dict[str, Any] + runtime: FeatureFlagsRuntimeClient, slug: str, context: Dict[str, Any] ) -> Any: - with_false = _is_enabled(configuration, slug, context, default=False) - with_true = _is_enabled(configuration, slug, context, default=True) + with_false = runtime.is_enabled(slug, context, default=False) + with_true = runtime.is_enabled(slug, context, default=True) if with_false == with_true: return with_false return "absent" if (with_false, with_true) == (False, True) else "inverted" @@ -63,15 +63,43 @@ def test_evaluation_corpus(case: Dict[str, Any]) -> None: configuration = FeatureFlagsConfiguration.from_dict( {"version": 2, "flags": {key: flag}} ) - for evaluation in case["evaluations"]: - fields = set(evaluation) - assert ( - _EVALUATION_FIELDS - <= fields - <= _EVALUATION_FIELDS | _OPTIONAL_EVALUATION_FIELDS - ), f"unexpected evaluation fields: {fields}" - observed = _observed(configuration, key, evaluation["context"]) - assert observed == evaluation["expect"], evaluation + with _runtime(configuration) as runtime: + for evaluation in case["evaluations"]: + fields = set(evaluation) + assert ( + _EVALUATION_FIELDS + <= fields + <= _EVALUATION_FIELDS | _OPTIONAL_EVALUATION_FIELDS + ), f"unexpected evaluation fields: {fields}" + observed = _observed(runtime, key, evaluation["context"]) + assert observed == evaluation["expect"], evaluation + + +def test_get_all_flags_agrees_with_is_enabled_and_omits_flags_without_an_answer() -> ( + None +): + flags = { + case.get("key", case["flag"]["slug"]) + str(index): { + **case["flag"], + "slug": case.get("key", case["flag"]["slug"]) + str(index), + } + for index, case in enumerate(EVALUATION["cases"]) + } + contexts: List[Dict[str, Any]] = [ + {}, + {"user": "user_1", "organization": "org_1"}, + {"organization": "org_acme"}, + ] + configuration = FeatureFlagsConfiguration.from_dict({"version": 2, "flags": flags}) + with _runtime(configuration) as runtime: + for context in contexts: + all_flags = runtime.get_all_flags(context) + for slug in flags: + answer = _observed(runtime, slug, context) + if answer == "absent": + assert slug not in all_flags + else: + assert all_flags[slug] is answer @pytest.mark.parametrize("case", PAYLOADS["cases"], ids=lambda case: case["name"]) @@ -106,10 +134,11 @@ def test_context_corpus_flag_matches_only_the_surviving_targets() -> None: configuration = FeatureFlagsConfiguration.from_dict( {"version": 2, "flags": {"flag": CONTEXTS["flag"]}} ) - for case in CONTEXTS["cases"]: - expected = ( - case["targets"].get("workspace") == "ws_1" - or case["targets"].get("user") == "user_1" - ) - observed = _is_enabled(configuration, "flag", case["context"], default=False) - assert observed is expected, case["name"] + with _runtime(configuration) as runtime: + for case in CONTEXTS["cases"]: + expected = ( + case["targets"].get("workspace") == "ws_1" + or case["targets"].get("user") == "user_1" + ) + observed = runtime.is_enabled("flag", case["context"]) + assert observed is expected, case["name"] diff --git a/tests/feature_flags_runtime/test_policy.py b/tests/feature_flags_runtime/test_policy.py new file mode 100644 index 00000000..56c6cd53 --- /dev/null +++ b/tests/feature_flags_runtime/test_policy.py @@ -0,0 +1,160 @@ +# @oagen-ignore-file +"""Polling delays and retry decisions of the runtime client.""" + +from __future__ import annotations + +from typing import Optional + +import pytest + +from workos._errors import ( + NotFoundError, + RateLimitExceededError, + ServerError, + WorkOSConnectionError, + WorkOSError, + WorkOSTimeoutError, +) +from workos._http import TransportError +from workos.feature_flags.runtime import _policy + + +@pytest.fixture +def random_value(monkeypatch: pytest.MonkeyPatch) -> "RandomValue": + value = RandomValue() + monkeypatch.setattr(_policy.random, "random", lambda: value.value) + return value + + +class RandomValue: + value = 0.5 + + +class TestNextDelay: + def test_a_healthy_poll_waits_one_interval(self, random_value: RandomValue) -> None: + assert _policy.next_delay(30, 0, 60) == 30 + + @pytest.mark.parametrize("value, expected", [(0.0, 27.0), (1.0, 33.0)]) + def test_jitter_is_ten_percent_either_way( + self, random_value: RandomValue, value: float, expected: float + ) -> None: + random_value.value = value + assert _policy.next_delay(30, 0, 60) == pytest.approx(expected) + + def test_failures_never_poll_sooner_than_the_interval( + self, random_value: RandomValue + ) -> None: + assert [_policy.next_delay(30, failures, 60) for failures in range(1, 9)] == [ + 30, + 30, + 30, + 30, + 30, + 32, + 60, + 60, + ] + + def test_backoff_is_capped_by_max_retry_delay( + self, random_value: RandomValue + ) -> None: + assert _policy.next_delay(5, 20, 12) == 12 + assert _policy.next_delay(5, 1000, 60) == 60 + + def test_delays_have_a_one_second_floor(self, random_value: RandomValue) -> None: + random_value.value = 0.0 + assert _policy.next_delay(1, 0, 60) == 1.0 + + +class TestAttemptRetryDelay: + @pytest.mark.parametrize( + "error", + [ + ServerError("busy", status_code=503), + RateLimitExceededError("slow down"), + WorkOSTimeoutError(), + WorkOSConnectionError(), + ], + ) + def test_transient_failures_are_retried( + self, random_value: RandomValue, error: Exception + ) -> None: + assert _policy.attempt_retry_delay(error, 0, 10, 3) == pytest.approx(0.5) + assert _policy.attempt_retry_delay(error, 2, 10, 3) == pytest.approx(1.125) + + def test_network_errors_from_the_transport_are_retried( + self, random_value: RandomValue + ) -> None: + error = WorkOSError("Network error") + error.__cause__ = TransportError("reset") + assert _policy.attempt_retry_delay(error, 0, 10, 3) is not None + + @pytest.mark.parametrize( + "error", + [ + NotFoundError("missing"), + ServerError("not implemented", status_code=501), + ValueError("bug"), + ], + ) + def test_other_failures_are_not_retried(self, error: Exception) -> None: + assert _policy.attempt_retry_delay(error, 0, 10, 3) is None + + @pytest.mark.parametrize("max_retries, attempt", [(3, 3), (1, 1), (0, 0)]) + def test_retries_follow_the_clients_max_retries( + self, max_retries: int, attempt: int + ) -> None: + error = ServerError("busy", status_code=503) + assert _policy.attempt_retry_delay(error, attempt, 10, max_retries) is None + if attempt: + assert _policy.attempt_retry_delay(error, attempt - 1, 10, max_retries) + + def test_retry_after_is_honored_when_it_fits(self) -> None: + error = RateLimitExceededError("slow down", retry_after=2.0) + assert _policy.attempt_retry_delay(error, 0, 10, 3) == 2.0 + assert _policy.attempt_retry_delay(error, 0, 1.5, 3) is None + + def test_retry_after_is_capped_and_never_negative_inside_a_refresh(self) -> None: + long_wait = RateLimitExceededError("slow down", retry_after=120.0) + assert _policy.attempt_retry_delay(long_wait, 0, 300, 3) == 60.0 + negative = RateLimitExceededError("slow down", retry_after=-5.0) + assert _policy.attempt_retry_delay(negative, 0, 10, 3) == 0.0 + + @pytest.mark.parametrize("remaining, expected", [(1.0, 0.5), (0.99, None)]) + def test_a_retry_must_leave_time_for_the_request_itself( + self, random_value: RandomValue, remaining: float, expected: Optional[float] + ) -> None: + assert ( + _policy.attempt_retry_delay( + ServerError("busy", status_code=503), 0, remaining, 3 + ) + == expected + ) + + +class TestDurations: + @pytest.mark.parametrize("value", [True, "10", None]) + def test_non_numbers_are_rejected(self, value: object) -> None: + with pytest.raises(TypeError): + _policy.validate_duration(value, "request_timeout") + + def test_the_polling_interval_has_a_five_second_minimum(self) -> None: + assert _policy.polling_interval(0.5) == 5.0 + assert _policy.polling_interval(45) == 45.0 + + +class TestRetryAfter: + def test_a_429_asks_the_next_poll_to_wait(self) -> None: + assert _policy.retry_after(RateLimitExceededError("slow", retry_after=20)) == 20 + + def test_the_wait_is_capped_at_sixty_seconds(self) -> None: + assert ( + _policy.retry_after(RateLimitExceededError("slow", retry_after=600)) == 60 + ) + + @pytest.mark.parametrize( + "error", + [RateLimitExceededError("slow"), ServerError("busy", status_code=503)], + ) + def test_other_failures_ask_for_nothing(self, error: Exception) -> None: + assert _policy.retry_after(error) is None diff --git a/tests/feature_flags_runtime/test_sync_client.py b/tests/feature_flags_runtime/test_sync_client.py new file mode 100644 index 00000000..48788be1 --- /dev/null +++ b/tests/feature_flags_runtime/test_sync_client.py @@ -0,0 +1,865 @@ +# @oagen-ignore-file +"""FeatureFlagsRuntimeClient: polling, readiness, notifications and shutdown.""" + +from __future__ import annotations + +import json +import logging +import threading +import traceback +from typing import Any, Callable, List + +import pytest + +import httpx2 + +from workos import HTTPResponse, TransportTimeout, WorkOSClient, create_public_client +from workos._errors import ( + AuthenticationError, + AuthorizationError, + ConfigurationError, + NotFoundError, + ServerError, +) +from workos.feature_flags.runtime import _configuration +from workos.feature_flags.runtime import ( + FeatureFlagsConfiguration, + FeatureFlagsConfigurationError, + FeatureFlagsRuntimeClient, + FlagsChangedEvent, + RuntimeStatus, +) + +from tests.feature_flags_runtime.helpers import ( + API_KEY, + BASE_URL, + Collector, + FlagServer, + SyncBackend, + eventually, + flag, + payload, + rule, +) + +Make = Callable[..., FeatureFlagsRuntimeClient] + +ORDERED = payload( + ordered=flag( + "ordered", + default_value=True, + rules=[ + rule("flag_rule_1", False, "organization", "org_1"), + rule("flag_rule_2", True, "user", "user_1"), + ], + ), + workspace=flag("workspace", rules=[rule("flag_rule_3", True, "workspace", "ws_1")]), +) +CHANGED = payload( + ordered=flag("ordered", default_value=False), + added=flag("added", default_value=True), +) + + +class TestRequests: + def test_polls_the_v2_endpoint_with_the_sdk_transport( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.respond(200, ORDERED) + runtime = make_runtime(request_timeout=7) + runtime.start() + runtime.wait_until_ready(timeout=5) + + request = server.requests[0] + assert request.method == "GET" + assert request.url == f"{BASE_URL}/sdk/feature-flags?payload_version=2" + assert request.headers["Authorization"] == f"Bearer {API_KEY}" + assert "Python SDK/" in request.headers["User-Agent"] + assert 0 < request.timeout <= 7 + + def test_makes_no_request_until_started( + self, server: FlagServer, make_runtime: Make + ) -> None: + runtime = make_runtime() + assert runtime.is_enabled("ordered") is False + assert server.requests == [] + + def test_start_is_idempotent( + self, server: FlagServer, make_runtime: Make, fast_polling: List[int] + ) -> None: + server.respond(200, ORDERED) + runtime = make_runtime() + runtime.start() + runtime.start() + server.wait_for_requests(3) + assert server.max_in_flight == 1 + + +class TestReadiness: + def test_ready_after_the_first_successful_poll( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.respond(200, ORDERED) + runtime = make_runtime() + assert runtime.is_ready is False + runtime.start() + runtime.wait_until_ready(timeout=5) + + assert runtime.is_ready + assert ( + runtime.is_enabled("ordered", {"user": "user_1", "organization": "org_1"}) + is False + ) + assert runtime.is_enabled("ordered", {"user": "user_1"}) is True + assert runtime.get_all_flags({"workspace": "ws_1"}) == { + "ordered": True, + "workspace": True, + } + status = runtime.status + assert status.flag_count == 2 + assert status.refresh_count == 1 + assert status.last_successful_refresh_at is not None + assert status.cache_age() is not None + + def test_bootstrap_is_ready_immediately_without_a_request( + self, server: FlagServer, make_runtime: Make + ) -> None: + bootstrap = FeatureFlagsConfiguration.from_dict(ORDERED) + runtime = make_runtime(bootstrap=bootstrap) + runtime.wait_until_ready(timeout=0) + + assert runtime.is_enabled("ordered") is True + assert runtime.status.last_successful_refresh_at is None + assert runtime.status.cache_age() is None + assert runtime.configuration is bootstrap + assert server.requests == [] + + def test_wait_times_out_without_a_configuration(self, make_runtime: Make) -> None: + runtime = make_runtime() + with pytest.raises(TimeoutError): + runtime.wait_until_ready(timeout=0.01) + + def test_unready_client_serves_caller_defaults(self, make_runtime: Make) -> None: + runtime = make_runtime() + assert runtime.is_enabled("ordered", default=True) is True + assert runtime.get_all_flags() == {} + assert runtime.get_flag("ordered") is None + + def test_a_401_before_readiness_fails_the_wait( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.respond(401, {"message": "Unauthorized"}) + runtime = make_runtime() + runtime.start() + with pytest.raises(AuthenticationError): + runtime.wait_until_ready(timeout=5) + assert runtime.status.polling_stopped + assert len(server.requests) == 1 + + def test_a_wait_timeout_names_the_last_refresh_error( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.respond(404, {"message": "Not Found"}) + runtime = make_runtime() + with pytest.raises(NotFoundError): + runtime.refresh() + with pytest.raises( + TimeoutError, match="the last refresh failed: NotFoundError" + ) as raised: + runtime.wait_until_ready(timeout=0.01) + assert raised.value.__cause__ is runtime.status.last_error + + def test_each_wait_after_a_401_raises_a_fresh_error( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.respond(401, {"message": "Unauthorized"}) + runtime = make_runtime() + with pytest.raises(AuthenticationError): + runtime.refresh() + errors: List[BaseException] = [] + + def wait() -> None: + try: + runtime.wait_until_ready(timeout=1) + except BaseException as error: + errors.append(error) + + for _ in range(3): + wait() + waiters = [threading.Thread(target=wait) for _ in range(3)] + for waiter in waiters: + waiter.start() + for waiter in waiters: + waiter.join(5) + + assert len(errors) == 6 + assert all(isinstance(error, AuthenticationError) for error in errors) + assert len({id(error) for error in errors}) == 6 + stored = errors[0].__cause__ + assert stored is not None and all(error.__cause__ is stored for error in errors) + assert all(error.status_code == 401 for error in errors) # type: ignore[attr-defined] + depths = {len(traceback.extract_tb(error.__traceback__)) for error in errors} + assert len(depths) == 1 + + def test_closing_before_readiness_wakes_waiters(self, make_runtime: Make) -> None: + runtime = make_runtime() + errors: List[BaseException] = [] + + def wait() -> None: + try: + runtime.wait_until_ready(timeout=5) + except BaseException as error: + errors.append(error) + + waiter = threading.Thread(target=wait) + waiter.start() + runtime.close() + waiter.join(5) + assert len(errors) == 1 and isinstance(errors[0], RuntimeError) + + +class TestChangeNotifications: + def test_first_load_reports_every_flag( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.respond(200, ORDERED) + changes = Collector() + runtime = make_runtime(on_change=changes) + runtime.start() + + (event,) = changes.wait_for(1) + assert isinstance(event, FlagsChangedEvent) + assert event.previous is None + assert event.changed_flags == {"ordered", "workspace"} + assert event.current is runtime.configuration + + def test_reports_only_real_changes( + self, server: FlagServer, make_runtime: Make, fast_polling: List[int] + ) -> None: + server.respond(200, ORDERED) + changes = Collector() + runtime = make_runtime() + runtime.add_change_listener(changes) + runtime.start() + changes.wait_for(1) + server.wait_for_requests(4) + assert len(changes.items) == 1 + + server.respond(200, CHANGED) + events = changes.wait_for(2) + assert events[1].changed_flags == {"ordered", "workspace", "added"} + assert events[1].previous is not None + assert events[1].previous.to_json() == events[0].current.to_json() + assert runtime.is_enabled("added") is True + assert runtime.get_flag("workspace") is None + + def test_first_refresh_after_bootstrap_compares_with_the_bootstrap( + self, server: FlagServer, make_runtime: Make + ) -> None: + bootstrap = FeatureFlagsConfiguration.from_dict(ORDERED) + changes = Collector() + statuses = Collector() + server.respond(200, CHANGED) + runtime = make_runtime( + bootstrap=bootstrap, on_change=changes, on_status=statuses + ) + runtime.refresh() + + (event,) = changes.wait_for(1) + assert event.previous is bootstrap + assert event.changed_flags == {"ordered", "workspace", "added"} + + runtime.refresh() + assert len(changes.items) == 1 + assert len(statuses.items) == 2 + + def test_an_unchanged_payload_keeps_the_snapshot_without_parsing_it( + self, server: FlagServer, make_runtime: Make, monkeypatch: pytest.MonkeyPatch + ) -> None: + server.respond(200, ORDERED) + changes = Collector() + runtime = make_runtime(on_change=changes) + runtime.refresh() + snapshot = runtime.configuration + first_refresh = runtime.status.last_successful_refresh_at + parses: List[object] = [] + original = _configuration._parse_payload + monkeypatch.setattr( + _configuration, + "_parse_payload", + lambda payload: parses.append(payload) or original(payload), + ) + # Same content, different key order. + server.respond(200, json.loads(json.dumps(ORDERED, sort_keys=True))) + runtime.refresh() + + assert runtime.configuration is snapshot + assert parses == [] + assert runtime.status.refresh_count == 2 + assert runtime.status.last_successful_refresh_at != first_refresh + assert len(changes.items) == 1 + + server.respond(200, CHANGED) + runtime.refresh() + assert len(parses) == 1 + assert runtime.configuration is not snapshot + assert len(changes.items) == 2 + + def test_an_identical_bootstrap_reports_no_change( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.respond(200, ORDERED) + changes = Collector() + runtime = make_runtime( + bootstrap=FeatureFlagsConfiguration.from_dict(ORDERED), on_change=changes + ) + runtime.refresh() + assert changes.items == [] + + def test_a_change_outside_the_flags_is_reported_for_persisted_caches( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.respond(200, ORDERED) + changes = Collector() + runtime = make_runtime(on_change=changes) + runtime.refresh() + server.respond(200, {**ORDERED, "served_by": "a newer server"}) + runtime.refresh() + + event = changes.wait_for(2)[1] + assert event.changed_flags == frozenset() + assert event.current.to_dict()["served_by"] == "a newer server" + + def test_listeners_can_refresh_and_events_stay_in_refresh_order( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.respond(200, ORDERED) + log: List[str] = [] + holder: List[FeatureFlagsRuntimeClient] = [] + refreshed = threading.Event() + + def on_change(event: FlagsChangedEvent) -> None: + label = "first-load" if event.previous is None else "refresh" + log.append(f"start {label}") + if event.previous is None: + # Refresh from another thread and wait for it, which needs the + # refresh lock to be free while listeners run. + server.respond(200, CHANGED) + refresher = threading.Thread(target=holder[0].refresh) + refresher.start() + refresher.join(5) + if not refresher.is_alive(): + refreshed.set() + log.append(f"end {label}") + + runtime = make_runtime(on_change=on_change) + holder.append(runtime) + runtime.refresh() + + assert refreshed.is_set() + eventually(lambda: len(log) == 4) + assert log == [ + "start first-load", + "end first-load", + "start refresh", + "end refresh", + ] + + def test_listener_failures_are_logged_and_do_not_stop_polling( + self, + server: FlagServer, + make_runtime: Make, + fast_polling: List[int], + caplog: pytest.LogCaptureFixture, + ) -> None: + server.respond(200, ORDERED) + later = Collector() + + def broken(event: FlagsChangedEvent) -> None: + raise RuntimeError("listener bug") + + runtime = make_runtime(on_change=broken) + runtime.add_change_listener(later) + with caplog.at_level(logging.ERROR, logger="workos.feature_flags.runtime"): + runtime.start() + later.wait_for(1) + server.respond(200, CHANGED) + later.wait_for(2) + assert "listener raised" in caplog.text + + def test_removed_listeners_are_not_called( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.respond(200, ORDERED) + changes = Collector() + runtime = make_runtime(on_change=changes) + runtime.remove_change_listener(changes) + runtime.refresh() + assert changes.items == [] + + def test_async_listeners_are_rejected_with_a_warning( + self, server: FlagServer, make_runtime: Make, caplog: pytest.LogCaptureFixture + ) -> None: + server.respond(200, ORDERED) + + async def listener(event: FlagsChangedEvent) -> None: + raise AssertionError("never run") + + runtime = make_runtime() + runtime.add_change_listener(listener) # type: ignore[arg-type] + with caplog.at_level(logging.WARNING, logger="workos.feature_flags.runtime"): + runtime.refresh() + assert "async listener" in caplog.text + + +class TestFailures: + def test_transient_failures_are_retried_inside_one_refresh( + self, server: FlagServer, make_runtime: Make, fast_polling: List[int] + ) -> None: + server.enqueue((503, {"message": "busy"}), (500, None)) + server.respond(200, ORDERED) + runtime = make_runtime() + runtime.refresh() + + assert len(server.requests) == 3 + assert runtime.status.refresh_count == 1 + assert runtime.status.failed_refresh_count == 0 + + def test_retries_stop_when_the_request_timeout_is_spent( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.respond(503, {"message": "busy"}) + runtime = make_runtime(request_timeout=0.05) + with pytest.raises(ServerError): + runtime.refresh() + # The shortest retry wait (0.25 s) does not fit in a 50 ms budget. + assert len(server.requests) == 1 + + @pytest.mark.parametrize("max_retries, requests", [(0, 1), (1, 2)]) + def test_retries_follow_the_clients_max_retries( + self, + server: FlagServer, + fast_polling: List[int], + max_retries: int, + requests: int, + ) -> None: + server.respond(503, {"message": "busy"}) + workos = WorkOSClient( + api_key=API_KEY, + base_url=BASE_URL, + http_client=SyncBackend(server), + max_retries=max_retries, + ) + runtime = workos.create_feature_flags_runtime_client() + with pytest.raises(ServerError): + runtime.refresh() + assert len(server.requests) == requests + runtime.close() + workos.close() + + def test_a_slow_failing_server_surfaces_its_error_not_a_timeout( + self, fast_polling: List[int] + ) -> None: + class SlowFailingBackend: + """Answers 503 after 0.3 s, and times out when given less than that.""" + + def __init__(self) -> None: + self.timeouts: List[float] = [] + + def request(self, method: str, url: str, **kwargs: Any) -> HTTPResponse: + self.timeouts.append(kwargs["timeout"]) + if kwargs["timeout"] < 0.3: + threading.Event().wait(kwargs["timeout"]) + raise TransportTimeout("read timed out") + threading.Event().wait(0.3) + return HTTPResponse(503, httpx2.Headers(), b"", method, url) + + def close(self) -> None: + pass + + backend = SlowFailingBackend() + workos = WorkOSClient(api_key=API_KEY, base_url=BASE_URL, http_client=backend) + runtime = workos.create_feature_flags_runtime_client(request_timeout=1.0) + with pytest.raises(ServerError): + runtime.refresh() + assert all(timeout >= 0.3 for timeout in backend.timeouts) + runtime.close() + workos.close() + + def test_a_429_retry_after_delays_the_next_poll( + self, server: FlagServer, make_runtime: Make, fast_polling: List[int] + ) -> None: + # Retry-After does not fit in the refresh budget, so the refresh fails; + # without honoring it the next poll would follow 10 ms later. + server.enqueue((429, {"message": "slow down"}, {"Retry-After": "30"})) + server.respond(200, ORDERED) + runtime = make_runtime(request_timeout=1) + runtime.start() + server.wait_for_requests(1) + threading.Event().wait(0.3) + assert len(server.requests) == 1 + assert runtime.status.failed_refresh_count == 1 + + def test_errors_that_retrying_cannot_fix_are_not_retried( + self, server: FlagServer, make_runtime: Make, fast_polling: List[int] + ) -> None: + server.respond(404, {"message": "Not Found"}) + runtime = make_runtime() + with pytest.raises(NotFoundError): + runtime.refresh() + assert len(server.requests) == 1 + + @pytest.mark.parametrize( + "status, body, error_type", + [ + (404, {"message": "Not Found"}, NotFoundError), + (403, {"message": "Forbidden"}, AuthorizationError), + (500, {"message": "Boom"}, ServerError), + (200, {"version": 3, "flags": {}}, FeatureFlagsConfigurationError), + (200, b"not json", FeatureFlagsConfigurationError), + ], + ) + def test_background_failures_keep_the_last_good_flags_and_recover( + self, + server: FlagServer, + make_runtime: Make, + fast_polling: List[int], + status: int, + body: Any, + error_type: type, + ) -> None: + server.respond(200, ORDERED) + errors = Collector() + changes = Collector() + runtime = make_runtime(on_error=errors, on_change=changes) + runtime.start() + changes.wait_for(1) + good = runtime.configuration + + server.respond(status, body) + (error, *_) = errors.wait_for(1) + assert isinstance(error, error_type) + assert runtime.configuration is good + assert runtime.is_enabled("ordered", {"user": "user_1"}) is True + assert isinstance(runtime.status.last_error, error_type) + assert runtime.status.failed_refresh_count >= 1 + + server.respond(200, CHANGED) + changes.wait_for(2) + eventually(lambda: runtime.status.last_error is None) + assert runtime.status.polling_stopped is False + + def test_failures_back_off_and_reset_after_success( + self, server: FlagServer, make_runtime: Make, fast_polling: List[int] + ) -> None: + server.enqueue(*[(404, {"message": "Not Found"})] * 3) + server.respond(200, ORDERED) + runtime = make_runtime() + runtime.start() + server.wait_for_requests(5) + assert fast_polling[:4] == [1, 2, 3, 0] + + def test_a_401_stops_polling_and_keeps_the_last_good_flags( + self, server: FlagServer, make_runtime: Make, fast_polling: List[int] + ) -> None: + server.respond(200, ORDERED) + errors = Collector() + statuses = Collector() + runtime = make_runtime(on_error=errors, on_status=statuses) + runtime.start() + runtime.wait_until_ready(timeout=5) + + server.respond(401, {"message": "Unauthorized"}) + (error,) = errors.wait_for(1) + assert isinstance(error, AuthenticationError) + eventually(lambda: runtime.status.polling_stopped) + sent = len(server.requests) + threading.Event().wait(0.1) + assert len(server.requests) == sent + assert runtime.is_enabled("ordered", {"user": "user_1"}) is True + runtime.wait_until_ready(timeout=0) + last: RuntimeStatus = statuses.items[-1] + assert last.polling_stopped and isinstance(last.last_error, AuthenticationError) + + def test_a_401_from_refresh_stops_polling( + self, server: FlagServer, make_runtime: Make, fast_polling: List[int] + ) -> None: + server.respond(200, ORDERED) + errors = Collector() + statuses = Collector() + runtime = make_runtime(on_error=errors, on_status=statuses) + runtime.start() + server.wait_for_requests(2) + + server.respond(401, {"message": "Unauthorized"}) + with pytest.raises(AuthenticationError): + runtime.refresh() + assert runtime.status.polling_stopped + sent = len(server.requests) + threading.Event().wait(0.1) + assert len(server.requests) == sent + assert errors.items == [] + assert statuses.items[-1].polling_stopped + assert runtime.is_enabled("ordered", {"user": "user_1"}) is True + + def test_a_401_from_refresh_before_start_fails_readiness_and_start( + self, server: FlagServer, make_runtime: Make, caplog: pytest.LogCaptureFixture + ) -> None: + server.respond(401, {"message": "Unauthorized"}) + runtime = make_runtime() + with pytest.raises(AuthenticationError): + runtime.refresh() + with pytest.raises(AuthenticationError): + runtime.wait_until_ready(timeout=1) + with caplog.at_level(logging.WARNING, logger="workos.feature_flags.runtime"): + runtime.start() + assert "Not starting feature flag polling" in caplog.text + threading.Event().wait(0.05) + assert len(server.requests) == 1 + + def test_failure_logs_carry_the_http_status_of_an_empty_body( + self, + server: FlagServer, + make_runtime: Make, + fast_polling: List[int], + caplog: pytest.LogCaptureFixture, + ) -> None: + server.respond(500, None) + with caplog.at_level(logging.WARNING, logger="workos.feature_flags.runtime"): + runtime = make_runtime() + runtime.start() + eventually(lambda: "ServerError: HTTP 500" in caplog.text) + assert "message=" not in caplog.text + + def test_a_403_keeps_polling( + self, server: FlagServer, make_runtime: Make, fast_polling: List[int] + ) -> None: + server.respond(403, {"message": "Forbidden"}) + runtime = make_runtime() + runtime.start() + server.wait_for_requests(3) + assert runtime.status.polling_stopped is False + + def test_unobserved_failures_are_logged( + self, + server: FlagServer, + make_runtime: Make, + fast_polling: List[int], + caplog: pytest.LogCaptureFixture, + ) -> None: + server.respond(500, {"message": "Boom"}) + with caplog.at_level(logging.WARNING, logger="workos.feature_flags.runtime"): + runtime = make_runtime() + runtime.start() + eventually( + lambda: "keeping the last good flags and retrying" in caplog.text + ) + + +class TestManualRefresh: + def test_works_without_polling( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.respond(200, ORDERED) + runtime = make_runtime() + runtime.refresh() + assert runtime.is_ready + assert len(server.requests) == 1 + + def test_raises_without_notifying_error_listeners( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.respond(404, {"message": "Not Found"}) + errors = Collector() + runtime = make_runtime(on_error=errors) + with pytest.raises(NotFoundError): + runtime.refresh() + assert errors.items == [] + assert runtime.status.failed_refresh_count == 1 + + def test_is_serialized_with_polling( + self, server: FlagServer, make_runtime: Make, fast_polling: List[int] + ) -> None: + server.respond(200, ORDERED) + runtime = make_runtime() + runtime.start() + refreshers = [threading.Thread(target=runtime.refresh) for _ in range(4)] + for refresher in refreshers: + refresher.start() + for refresher in refreshers: + refresher.join(5) + assert server.max_in_flight == 1 + + +class TestClose: + def test_stops_polling_and_keeps_serving_the_last_flags( + self, server: FlagServer, make_runtime: Make, fast_polling: List[int] + ) -> None: + server.respond(200, ORDERED) + runtime = make_runtime() + runtime.start() + server.wait_for_requests(2) + runtime.close() + sent = len(server.requests) + threading.Event().wait(0.1) + + assert len(server.requests) == sent + assert runtime.is_closed + assert runtime.is_enabled("ordered", {"user": "user_1"}) is True + with pytest.raises(RuntimeError): + runtime.start() + with pytest.raises(RuntimeError): + runtime.refresh() + runtime.close() + + def test_discards_a_refresh_in_flight( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.enqueue(FlagServer.BLOCK) + changes = Collector() + runtime = make_runtime(on_change=changes) + runtime.start() + server.wait_for_requests(1) + closer = threading.Thread(target=runtime.close) + closer.start() + eventually(lambda: runtime.is_closed) + server.release() + closer.join(5) + + assert not closer.is_alive() + assert runtime.is_ready is False + assert changes.items == [] + + def test_every_caller_waits_for_the_poller_to_stop( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.enqueue(FlagServer.BLOCK) + runtime = make_runtime() + runtime.start() + server.wait_for_requests(1) + first = threading.Thread(target=runtime.close) + first.start() + eventually(lambda: runtime.is_closed) + second = threading.Thread(target=runtime.close) + second.start() + second.join(0.2) + + assert second.is_alive() + server.release() + first.join(5) + second.join(5) + assert not second.is_alive() + assert runtime._thread is not None and not runtime._thread.is_alive() + + def test_returns_only_after_a_manual_refresh_finishes_installing( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.respond(200, ORDERED) + runtime = make_runtime() + install = runtime._record_success + returned_during_install: List[bool] = [] + + def install_while_closing(*args: Any, **kwargs: Any) -> Any: + # close() from another thread just as the refresh installs. + closer = threading.Thread(target=runtime.close) + closer.start() + closer.join(0.2) + returned_during_install.append(not closer.is_alive()) + return install(*args, **kwargs) + + runtime._record_success = install_while_closing # type: ignore[method-assign] + runtime.refresh() + + assert returned_during_install == [False] + eventually(lambda: runtime.is_closed) + assert runtime.is_enabled("ordered", {"user": "user_1"}) is True + + def test_can_be_called_from_a_listener( + self, server: FlagServer, make_runtime: Make + ) -> None: + server.respond(200, ORDERED) + closed = threading.Event() + holder: List[FeatureFlagsRuntimeClient] = [] + + def close_on_change(event: FlagsChangedEvent) -> None: + holder[0].close() + closed.set() + + runtime = make_runtime(on_change=close_on_change) + holder.append(runtime) + runtime.start() + assert closed.wait(5) + assert runtime.is_closed + + def test_context_manager_starts_and_closes( + self, server: FlagServer, client: WorkOSClient + ) -> None: + server.respond(200, ORDERED) + with client.create_feature_flags_runtime_client() as runtime: + runtime.wait_until_ready(timeout=5) + assert runtime.is_closed + + +class TestConcurrency: + def test_readers_always_see_one_complete_snapshot( + self, server: FlagServer, make_runtime: Make, fast_polling: List[int] + ) -> None: + # Every snapshot holds flags a and b with equal values, alternating + # between all-true and all-false; a torn read would mix them. + def snapshot(value: bool) -> dict: + return payload( + a=flag("a", default_value=value), b=flag("b", default_value=value) + ) + + server.respond(200, snapshot(True)) + runtime = make_runtime() + runtime.refresh() + runtime.start() + torn: List[dict] = [] + stop = threading.Event() + + def read() -> None: + while not stop.is_set(): + flags = runtime.get_all_flags() + if len(set(flags.values())) != 1: + torn.append(flags) + + readers = [threading.Thread(target=read) for _ in range(4)] + for reader in readers: + reader.start() + for index in range(20): + server.respond(200, snapshot(index % 2 == 0)) + server.wait_for_requests(len(server.requests) + 1) + stop.set() + for reader in readers: + reader.join(5) + assert torn == [] + + +class TestConfiguration: + def test_rejects_a_client_without_an_api_key(self) -> None: + public = create_public_client(client_id="client_123") + with pytest.raises(ConfigurationError): + FeatureFlagsRuntimeClient(public) # type: ignore[arg-type] + public.close() + + def test_rejects_an_unparsed_bootstrap(self, client: WorkOSClient) -> None: + with pytest.raises(TypeError, match="from_json"): + client.create_feature_flags_runtime_client(bootstrap=ORDERED) # type: ignore[arg-type] + + def test_clamps_the_polling_interval(self, make_runtime: Make) -> None: + assert make_runtime(polling_interval=1)._polling_interval == 5.0 + + @pytest.mark.parametrize("value", [0, -1, float("nan"), float("inf")]) + def test_rejects_unusable_durations( + self, client: WorkOSClient, value: float + ) -> None: + with pytest.raises(ValueError): + client.create_feature_flags_runtime_client(request_timeout=value) + + def test_constructs_directly(self, server: FlagServer) -> None: + workos = WorkOSClient(api_key=API_KEY, base_url=BASE_URL) + runtime = FeatureFlagsRuntimeClient(workos, polling_interval=60) + assert runtime._polling_interval == 60.0 + runtime.close() + workos.close() diff --git a/tests/feature_flags_runtime/test_transport.py b/tests/feature_flags_runtime/test_transport.py new file mode 100644 index 00000000..7c9dca72 --- /dev/null +++ b/tests/feature_flags_runtime/test_transport.py @@ -0,0 +1,88 @@ +# @oagen-ignore-file +"""The sync client's poll request over a real HTTP connection.""" + +from __future__ import annotations + +import time +from typing import Iterator + +import httpx2 +import pytest + +from workos import WorkOSClient +from workos._errors import WorkOSTimeoutError + +from tests.feature_flags_runtime.helpers import ( + API_KEY, + LocalHTTPServer, + flag, + payload, +) + +PAYLOAD = payload(checkout=flag("checkout", default_value=True)) + + +@pytest.fixture +def local() -> Iterator[LocalHTTPServer]: + server = LocalHTTPServer(PAYLOAD) + yield server + server.close() + + +def test_sends_the_sdk_url_and_headers(local: LocalHTTPServer) -> None: + workos = WorkOSClient(api_key=API_KEY, base_url=local.base_url) + runtime = workos.create_feature_flags_runtime_client() + runtime.refresh() + + request = local.requests[0] + assert request["path"] == "/sdk/feature-flags?payload_version=2" + assert request["headers"]["Authorization"] == f"Bearer {API_KEY}" + assert request["headers"]["User-Agent"].startswith("WorkOS Python/") + assert runtime.is_enabled("checkout") is True + + runtime.close() + workos.close() + + +def test_uses_a_caller_supplied_httpx_client(local: LocalHTTPServer) -> None: + http_client = httpx2.Client(headers={"X-Caller-Client": "yes"}) + workos = WorkOSClient( + api_key=API_KEY, base_url=local.base_url, http_client=http_client + ) + runtime = workos.create_feature_flags_runtime_client() + runtime.refresh() + + assert local.requests[0]["headers"]["X-Caller-Client"] == "yes" + runtime.close() + assert not http_client.is_closed + http_client.close() + + +@pytest.mark.parametrize( + "mode, setting", + [ + # Headers arrive late: each phase only gets the time left. + ("stall", 3.0), + # Every read succeeds well within its own timeout, but the body as a + # whole would take several seconds. + ("drip", 0.1), + ], +) +def test_a_refresh_is_bounded_by_its_request_timeout( + local: LocalHTTPServer, mode: str, setting: float +) -> None: + local.mode = mode + local.stall = setting + local.drip_interval = setting + workos = WorkOSClient(api_key=API_KEY, base_url=local.base_url, max_retries=0) + runtime = workos.create_feature_flags_runtime_client(request_timeout=0.5) + + started = time.monotonic() + with pytest.raises(WorkOSTimeoutError): + runtime.refresh() + elapsed = time.monotonic() - started + + # The documented bound is request_timeout plus at most one read. + assert elapsed < 1.2 + runtime.close() + workos.close() diff --git a/tests/smoke_test.py b/tests/smoke_test.py index e630a96f..03eee258 100644 --- a/tests/smoke_test.py +++ b/tests/smoke_test.py @@ -198,6 +198,24 @@ def test_pagination_importable() -> None: print("✓ Pagination and RequestOptions importable") +def test_feature_flags_runtime_client_importable() -> None: + """Verify the runtime client constructs without making a request.""" + from workos import ( + FeatureFlagsConfiguration, + FeatureFlagsRuntimeClient, + WorkOSClient, + ) + + configuration = FeatureFlagsConfiguration.from_dict({"version": 2, "flags": {}}) + sync_runtime = WorkOSClient( + api_key="sk_test_smoke" + ).create_feature_flags_runtime_client(bootstrap=configuration) + assert isinstance(sync_runtime, FeatureFlagsRuntimeClient) + assert sync_runtime.is_enabled("missing", default=True) is True + sync_runtime.close() + print("✓ Feature flags runtime client importable") + + def test_dependencies_available() -> None: """Verify core runtime dependencies are installed and importable.""" import cryptography @@ -225,6 +243,7 @@ def main() -> int: test_core_types_importable, test_errors_importable, test_pagination_importable, + test_feature_flags_runtime_client_importable, test_dependencies_available, ]