From a195491253cf0f0cafc31b848234bef3e9c69e80 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Fri, 30 Oct 2026 09:59:53 -0500 Subject: [PATCH 1/3] feat: Accept synchronous hooks on the async client AsyncLDClient now takes the same ldclient.hook.Hook as LDClient, so integrations such as the OpenTelemetry tracing hook work with it. The hook stages run synchronously in the caller's task, so a span made current in the before stage stays current until the after stage. AsyncHook is removed. AsyncPlugin keeps its async register(), and its get_hooks() returns Hook instances. AsyncConfig now raises TypeError for a hook that is not a Hook or a plugin that is not an AsyncPlugin, which matches add_hook(). --- contract-tests/async_client_entity.py | 17 +- contract-tests/hook.py | 62 ++---- ldclient/async_client.py | 40 ++-- ldclient/async_config.py | 30 ++- ldclient/hook.py | 65 +----- ldclient/impl/client_common.py | 6 +- ldclient/plugin.py | 9 +- ldclient/testing/test_async_client.py | 35 +-- ldclient/testing/test_async_hooks.py | 310 +++++++++++++------------- 9 files changed, 254 insertions(+), 320 deletions(-) diff --git a/contract-tests/async_client_entity.py b/contract-tests/async_client_entity.py index 2569a156..56a845b5 100644 --- a/contract-tests/async_client_entity.py +++ b/contract-tests/async_client_entity.py @@ -2,12 +2,13 @@ import json import logging import sys +from concurrent.futures import ThreadPoolExecutor from typing import Any, Callable, Optional import requests from async_big_segment_store_fixture import AsyncBigSegmentStoreFixture from async_flag_change_listener import AsyncListenerRegistry -from hook import AsyncPostingHook +from hook import QueuedPostingHook from ldclient import Context from ldclient.async_client import AsyncLDClient @@ -40,6 +41,7 @@ def __init__(self, tag: str, config_params: dict): self.log = logging.getLogger(tag) self._client: Optional[AsyncLDClient] = None self._listeners: Optional[AsyncListenerRegistry] = None + self._hook_executor: Optional[ThreadPoolExecutor] = None self._config_params = config_params self._tag = tag @@ -91,10 +93,11 @@ async def start(self): else: opts["send_events"] = False - hooks = [] if config_params.get("hooks") is not None: - hooks = [ - AsyncPostingHook(h["name"], h["callbackUri"], h.get("data", {}), h.get("errors", {})) + # One worker keeps the posts from all hooks of this client in order. + self._hook_executor = ThreadPoolExecutor(max_workers=1) + opts["hooks"] = [ + QueuedPostingHook(h["name"], h["callbackUri"], h.get("data", {}), h.get("errors", {}), self._hook_executor) for h in config_params["hooks"]["hooks"] ] @@ -115,10 +118,6 @@ async def start(self): sdk_config = AsyncConfig(**opts) self._client = AsyncLDClient(sdk_config) - # The async client accepts AsyncHook instances only; register the - # harness's async posting hooks via add_hook() before start(). - for hook in hooks: - self._client.add_hook(hook) await self._client.start(start_wait / 1000.0) self._listeners = AsyncListenerRegistry(self._client.flag_tracker) @@ -279,6 +278,8 @@ async def close(self): await self._listeners.close_all() if self._client is not None: await self._client.close() + if self._hook_executor is not None: + self._hook_executor.shutdown(wait=False) self.log.info('Test ended') diff --git a/contract-tests/hook.py b/contract-tests/hook.py index 25feac37..55c912d1 100644 --- a/contract-tests/hook.py +++ b/contract-tests/hook.py @@ -1,10 +1,10 @@ -import asyncio +from concurrent.futures import ThreadPoolExecutor from typing import Optional import requests from ldclient.evaluation import EvaluationDetail -from ldclient.hook import AsyncHook, EvaluationSeriesContext, Hook, Metadata +from ldclient.hook import EvaluationSeriesContext, Hook, Metadata class PostingHook(Hook): @@ -47,56 +47,24 @@ def __post(self, stage: str, series_context: EvaluationSeriesContext, data: dict 'reason': detail.reason, } - requests.post(self.__callback, json=payload) + self._send(self.__callback, payload) return {**data, **self.__data.get(stage, {})} + def _send(self, callback: str, payload: dict): + requests.post(callback, json=payload) -class AsyncPostingHook(AsyncHook): - """AsyncHook mirror of PostingHook for the async contract-test harness. - The blocking requests.post is offloaded to a worker thread so it does not - block the event loop. - """ - - def __init__(self, name: str, callback: str, data: dict, errors: dict): - self.__name = name - self.__callback = callback - self.__data = data - self.__errors = errors - - @property - def metadata(self) -> Metadata: - return Metadata(name=self.__name) - - async def before_evaluation(self, series_context: EvaluationSeriesContext, data: dict) -> dict: - return await self.__post("beforeEvaluation", series_context, data, None) +class QueuedPostingHook(PostingHook): + """PostingHook for the async client. - async def after_evaluation(self, series_context: EvaluationSeriesContext, data: dict, detail: EvaluationDetail) -> dict: - return await self.__post("afterEvaluation", series_context, data, detail) - - async def __post(self, stage: str, series_context: EvaluationSeriesContext, data: dict, detail: Optional[EvaluationDetail]) -> dict: - if stage in self.__errors: - raise Exception(self.__errors[stage]) - - payload = { - 'evaluationSeriesContext': { - 'flagKey': series_context.key, - 'context': series_context.context.to_dict(), - 'defaultValue': series_context.default_value, - 'method': series_context.method, - }, - 'evaluationSeriesData': data, - 'stage': stage, - } - - if detail is not None: - payload['evaluationDetail'] = { - 'value': detail.value, - 'variationIndex': detail.variation_index, - 'reason': detail.reason, - } + The async client runs hook stages on the event loop, so the hook must not + block. The posts go to a single worker thread, which sends them in order. + """ - await asyncio.to_thread(requests.post, self.__callback, json=payload) + def __init__(self, name: str, callback: str, data: dict, errors: dict, executor: ThreadPoolExecutor): + super().__init__(name, callback, data, errors) + self.__executor = executor - return {**data, **self.__data.get(stage, {})} + def _send(self, callback: str, payload: dict): + self.__executor.submit(requests.post, callback, json=payload) diff --git a/ldclient/async_client.py b/ldclient/async_client.py index 9e740b75..9c7fdebe 100644 --- a/ldclient/async_client.py +++ b/ldclient/async_client.py @@ -12,8 +12,8 @@ from ldclient.context import Context from ldclient.evaluation import EvaluationDetail, FeatureFlagsState from ldclient.hook import ( - AsyncHook, EvaluationSeriesContext, + Hook, _EvaluationWithHookResult ) from ldclient.impl import AnyNum @@ -86,7 +86,7 @@ def __init__(self, config: AsyncConfig): # drop events. self._event_processor: Any = AsyncNullEventProcessor() - self.__hooks: List = list(config.hooks) + self.__hooks: List[Hook] = list(config.hooks) self._event_factory_default = EventFactory(False) self._event_factory_with_reasons = EventFactory(True) @@ -639,20 +639,22 @@ def secure_mode_hash(self, context: Context) -> str: """ return _secure_mode_hash(self._config, context) - def add_hook(self, hook: AsyncHook): + def add_hook(self, hook: Hook): """ Add a hook to the client. In order to register a hook before the client starts, please use the `hooks` property of `AsyncConfig`. Hooks provide entrypoints which allow for observation of SDK functions. - The async client only accepts :class:`ldclient.hook.AsyncHook` instances; - passing a synchronous :class:`ldclient.hook.Hook` raises ``TypeError``. + The hook stages run synchronously on the event loop, between the + start and the end of each evaluation. They must not block. A hook that + must do I/O should put the work on a queue and do it in the background. :param hook: + :raises TypeError: if ``hook`` is not a :class:`ldclient.hook.Hook` """ - if not isinstance(hook, AsyncHook): - raise TypeError("AsyncLDClient requires an AsyncHook; synchronous Hook instances are not supported") + if not isinstance(hook, Hook): + raise TypeError(f"AsyncLDClient requires a Hook; got {type(hook).__name__}") self.__hooks.append(hook) @@ -668,33 +670,33 @@ async def __evaluate_with_hooks(self, key: str, context: Context, default_value: # :return: """ # Snapshot the hook list to ensure hooks added during evaluation don't get called for the current evaluation. - hooks = list(self.__hooks) # type: List[AsyncHook] + hooks = list(self.__hooks) if not hooks: return await block() + # The stages run in the caller's task, not in a new task or thread. A hook + # can then change context variables in the before stage (for example, to + # make an OpenTelemetry span current) and restore them in the after stage. series_context = EvaluationSeriesContext(key=key, context=context, default_value=default_value, method=method) - hook_data = await self.__execute_before_evaluation(hooks, series_context) + hook_data = self.__execute_before_evaluation(hooks, series_context) evaluation_result = await block() - await self.__execute_after_evaluation(hooks, series_context, hook_data, evaluation_result.evaluation_detail) + self.__execute_after_evaluation(hooks, series_context, hook_data, evaluation_result.evaluation_detail) return evaluation_result - async def __execute_before_evaluation(self, hooks: List[AsyncHook], series_context: EvaluationSeriesContext) -> List[dict]: - return [await self.__try_execute_stage("beforeEvaluation", hook.metadata.name, lambda: hook.before_evaluation(series_context, {})) for hook in hooks] + def __execute_before_evaluation(self, hooks: List[Hook], series_context: EvaluationSeriesContext) -> List[dict]: + return [self.__try_execute_stage("beforeEvaluation", hook.metadata.name, lambda: hook.before_evaluation(series_context, {})) for hook in hooks] - async def __execute_after_evaluation(self, hooks: List[AsyncHook], series_context: EvaluationSeriesContext, hook_data: List[dict], evaluation_detail: EvaluationDetail) -> List[dict]: + def __execute_after_evaluation(self, hooks: List[Hook], series_context: EvaluationSeriesContext, hook_data: List[dict], evaluation_detail: EvaluationDetail) -> List[dict]: return [ - await self.__try_execute_stage("afterEvaluation", hook.metadata.name, lambda: hook.after_evaluation(series_context, data, evaluation_detail)) + self.__try_execute_stage("afterEvaluation", hook.metadata.name, lambda: hook.after_evaluation(series_context, data, evaluation_detail)) for (hook, data) in reversed(list(zip(hooks, hook_data))) ] - async def __try_execute_stage(self, method: str, hook_name: str, block: Callable[[], Any]) -> dict: + def __try_execute_stage(self, method: str, hook_name: str, block: Callable[[], dict]) -> dict: try: - return await block() - except asyncio.CancelledError: - # Do not swallow cancellation; it must propagate for shutdown. - raise + return block() except BaseException as e: log.error(f"An error occurred in {method} of the hook {hook_name}: #{e}") return {} diff --git a/ldclient/async_config.py b/ldclient/async_config.py index 7e28a680..4681467a 100644 --- a/ldclient/async_config.py +++ b/ldclient/async_config.py @@ -25,7 +25,7 @@ HTTPConfig, PrivateAttributesConfig ) -from ldclient.hook import AsyncHook +from ldclient.hook import Hook from ldclient.impl.aio.concurrency import AsyncEvent from ldclient.impl.util import ( log, @@ -130,6 +130,15 @@ class AsyncDataSystemConfig: """An optional fallback synchronizer that will read from FDv1""" +def _require_instances(items: Optional[list], kind: type, name: str) -> list: + if not items: + return [] + for item in items: + if not isinstance(item, kind): + raise TypeError(f"AsyncConfig {name} must be {kind.__name__} instances; got {type(item).__name__}") + return list(items) + + class AsyncConfig(DataSourceBuilderConfig, PrivateAttributesConfig): """Advanced configuration options for the async SDK client. @@ -172,7 +181,7 @@ def __init__( http: HTTPConfig = HTTPConfig(), big_segments: Optional[AsyncBigSegmentsConfig] = None, application: Optional[dict] = None, - hooks: Optional[List[AsyncHook]] = None, + hooks: Optional[List[Hook]] = None, plugins: Optional[List[AsyncPlugin]] = None, enable_event_compression: bool = False, omit_anonymous_contexts: bool = False, @@ -243,8 +252,11 @@ def __init__( :param http: Optional properties for customizing the client's HTTP/HTTPS behavior. See :class:`ldclient.config.HTTPConfig`. :param application: Optional properties for setting application metadata. See :py:attr:`~application` - :param hooks: Hooks provide entrypoints which allow for observation of SDK functions. + :param hooks: Hooks provide entrypoints which allow for observation of SDK functions. The hook + stages run synchronously on the event loop and must not block. See :py:attr:`~hooks`. :param plugins: A list of plugins to be used with the SDK. Plugin support is currently experimental and subject to change. + :raises TypeError: if an item in ``hooks`` is not a :class:`ldclient.hook.Hook`, or an item in + ``plugins`` is not an :class:`ldclient.plugin.AsyncPlugin`. :param enable_event_compression: Whether or not to enable GZIP compression for outgoing events. :param omit_anonymous_contexts: Sets whether anonymous contexts should be omitted from index and identify events. :param payload_filter_key: The payload filter is used to selectively limited the flags and segments delivered in the data source payload. Payload filtering is not supported with the FDv2 data system, so this has no effect on FDv2 requests. @@ -281,8 +293,8 @@ def __init__( self.__http = http self.__big_segments = AsyncBigSegmentsConfig() if not big_segments else big_segments self.__application = validate_application_info(application or {}, log) - self.__hooks = [hook for hook in hooks if isinstance(hook, AsyncHook)] if hooks else [] - self.__plugins = [plugin for plugin in plugins if isinstance(plugin, AsyncPlugin)] if plugins else [] + self.__hooks = _require_instances(hooks, Hook, 'hooks') + self.__plugins = _require_instances(plugins, AsyncPlugin, 'plugins') self.__enable_event_compression = enable_event_compression self.__omit_anonymous_contexts = omit_anonymous_contexts self.__payload_filter_key = payload_filter_key @@ -429,12 +441,18 @@ def application(self) -> dict: return self.__application @property - def hooks(self) -> List[AsyncHook]: + def hooks(self) -> List[Hook]: """ Initial set of hooks for the client. Hooks provide entrypoints which allow for observation of SDK functions. + The async client uses the same :class:`ldclient.hook.Hook` type as the + sync client. The hook stages run synchronously on the event loop, + between the start and the end of each evaluation. They must not block. + A hook that must do I/O should put the work on a queue and do it in the + background. + LaunchDarkly provides integration packages, and most applications will not need to implement their own hooks. """ diff --git a/ldclient/hook.py b/ldclient/hook.py index 49ddba8d..e4ebb371 100644 --- a/ldclient/hook.py +++ b/ldclient/hook.py @@ -38,6 +38,12 @@ class Hook(ABC): This class includes default implementations for all hook handlers. This allows LaunchDarkly to expand the list of hook handlers without breaking customer integrations. + + Both :class:`ldclient.client.LDClient` and + :class:`ldclient.async_client.AsyncLDClient` use this class. The async + client calls the handlers synchronously on the event loop, so a handler + must not block. A hook that must do I/O should put the work on a queue and + do it in the background. """ @property @@ -80,65 +86,6 @@ def after_evaluation(self, series_context: EvaluationSeriesContext, data: dict, return data -class AsyncHook(ABC): - """ - Abstract class for extending AsyncLDClient functionality via hooks. - - .. caution:: - This feature is experimental and should NOT be considered ready for production - use. It may change or be removed without notice and is not subject to backwards - compatibility guarantees. Pin to a specific minor version and review the changelog - before upgrading. - - All provided async hook implementations **MUST** inherit from this class. - - This class includes default implementations for all hook handlers. This - allows LaunchDarkly to expand the list of hook handlers without breaking - customer integrations. - - Unlike :class:`Hook`, the before and after methods are coroutines and will - be awaited by the async client. - """ - - @property - @abstractmethod - def metadata(self) -> Metadata: - """ - Get metadata about the hook implementation. - """ - return Metadata(name='UNDEFINED') - - async def before_evaluation(self, series_context: EvaluationSeriesContext, data: dict) -> dict: - """ - The before method is called during the execution of a variation method - before the flag value has been determined. The method is a coroutine - and will be awaited. - - :param series_context: Contains information about the evaluation being performed. This is not mutable. - :param data: A record associated with each stage of hook invocations. - Each stage is called with the data of the previous stage for a series. - The input record should not be modified. - :return: Data to use when executing the next state of the hook in the evaluation series. - """ - return data - - async def after_evaluation(self, series_context: EvaluationSeriesContext, data: dict, - detail: EvaluationDetail) -> dict: - """ - The after method is called during the execution of the variation method - after the flag value has been determined. The method is a coroutine - and will be awaited. - - :param series_context: Contains read-only information about the - evaluation being performed. - :param data: A record associated with each stage of hook invocations. - Each stage is called with the data of the previous stage for a series. - :param detail: The result of the evaluation. This value should not be modified. - :return: Data to use when executing the next state of the hook in the evaluation series. - """ - return data - - @dataclass class _EvaluationWithHookResult: evaluation_detail: EvaluationDetail diff --git a/ldclient/impl/client_common.py b/ldclient/impl/client_common.py index a99b5eba..6ba0fc07 100644 --- a/ldclient/impl/client_common.py +++ b/ldclient/impl/client_common.py @@ -14,7 +14,7 @@ from ldclient.config import SdkIdentityConfig from ldclient.context import Context -from ldclient.hook import AsyncHook, Hook +from ldclient.hook import Hook from ldclient.impl.util import log from ldclient.plugin import ( ApplicationMetadata, @@ -48,8 +48,8 @@ def get_environment_metadata(config: SdkIdentityConfig, sdk_name: str) -> Enviro ) -def get_plugin_hooks(plugins: Sequence[Union[Plugin, AsyncPlugin]], environment_metadata: EnvironmentMetadata) -> List: - hooks: List = [] +def get_plugin_hooks(plugins: Sequence[Union[Plugin, AsyncPlugin]], environment_metadata: EnvironmentMetadata) -> List[Hook]: + hooks: List[Hook] = [] for plugin in plugins: try: hooks.extend(plugin.get_hooks(environment_metadata)) diff --git a/ldclient/plugin.py b/ldclient/plugin.py index 461d361a..06986d32 100644 --- a/ldclient/plugin.py +++ b/ldclient/plugin.py @@ -6,7 +6,7 @@ from ldclient.context import Context from ldclient.evaluation import EvaluationDetail, FeatureFlagsState -from ldclient.hook import AsyncHook, Hook +from ldclient.hook import Hook from ldclient.impl import AnyNum from ldclient.impl.evaluator import error_reason from ldclient.interfaces import ( @@ -156,15 +156,16 @@ async def register(self, client: 'AsyncLDClient', metadata: EnvironmentMetadata) """ pass - def get_hooks(self, metadata: EnvironmentMetadata) -> List[AsyncHook]: + def get_hooks(self, metadata: EnvironmentMetadata) -> List[Hook]: """ Get a list of hooks that this plugin provides. This method is called before register() to collect all hooks from plugins. The hooks returned will be added to the SDK's hook configuration. - Async plugins provide async :class:`AsyncHook` instances only. + The async client uses the same :class:`ldclient.hook.Hook` type as the + sync client. - This method is synchronous (returns a list immediately — no I/O). + This method is synchronous and must not do I/O. :param metadata: Metadata about the environment in which the SDK is running :return: A list of hooks to be registered with the SDK diff --git a/ldclient/testing/test_async_client.py b/ldclient/testing/test_async_client.py index 5d3f17aa..02857e9b 100644 --- a/ldclient/testing/test_async_client.py +++ b/ldclient/testing/test_async_client.py @@ -169,9 +169,9 @@ async def test_migration_variation_returns_default_stage(): @pytest.mark.asyncio async def test_hooks_are_invoked_during_variation(): """Hooks added via add_hook() have before/after called during variation().""" - from ldclient.hook import AsyncHook, Metadata + from ldclient.hook import Hook, Metadata - class RecordingHook(AsyncHook): + class RecordingHook(Hook): def __init__(self): self.before_calls = [] self.after_calls = [] @@ -180,11 +180,11 @@ def __init__(self): def metadata(self): return Metadata(name='recording-hook') - async def before_evaluation(self, series_context, data): + def before_evaluation(self, series_context, data): self.before_calls.append(series_context) return data - async def after_evaluation(self, series_context, data, detail): + def after_evaluation(self, series_context, data, detail): self.after_calls.append((series_context, detail)) return data @@ -204,24 +204,11 @@ async def after_evaluation(self, series_context, data, detail): @pytest.mark.asyncio -async def test_add_hook_rejects_sync_hook(): - """add_hook() raises TypeError when given a synchronous Hook.""" - from ldclient.hook import EvaluationSeriesContext, Hook, Metadata - - class SyncHook(Hook): - @property - def metadata(self): - return Metadata(name='sync-hook') - - def before_evaluation(self, series_context: EvaluationSeriesContext, data: dict) -> dict: - return data - - def after_evaluation(self, series_context, data, detail): - return data - +async def test_add_hook_rejects_non_hook(): + """add_hook() raises TypeError when given an object that is not a Hook.""" client = AsyncLDClient(_offline_config()) with pytest.raises(TypeError): - client.add_hook(SyncHook()) + client.add_hook(object()) # type: ignore[arg-type] @pytest.mark.asyncio @@ -323,11 +310,11 @@ async def test_is_offline_reflects_config(): @pytest.mark.asyncio async def test_hooks_data_isolation(): """Each hook's before_evaluation receives its own isolated {} — not data from a prior hook.""" - from ldclient.hook import AsyncHook, Metadata + from ldclient.hook import Hook, Metadata received_data_by_hook = {} - class IsolationHook(AsyncHook): + class IsolationHook(Hook): def __init__(self, name, inject_key=None, inject_val=None): self._name = name self._inject_key = inject_key @@ -337,14 +324,14 @@ def __init__(self, name, inject_key=None, inject_val=None): def metadata(self): return Metadata(name=self._name) - async def before_evaluation(self, series_context, data): + def before_evaluation(self, series_context, data): # Record a copy of what we received received_data_by_hook[self._name] = dict(data) if self._inject_key: data[self._inject_key] = self._inject_val return data - async def after_evaluation(self, series_context, data, detail): + def after_evaluation(self, series_context, data, detail): return data hook_a = IsolationHook('hook-a', inject_key='hook_a', inject_val=True) diff --git a/ldclient/testing/test_async_hooks.py b/ldclient/testing/test_async_hooks.py index cd007b37..8cbb2dec 100644 --- a/ldclient/testing/test_async_hooks.py +++ b/ldclient/testing/test_async_hooks.py @@ -1,199 +1,209 @@ """ -Tests for AsyncHook dispatch in the evaluation pipeline. - -The _evaluate_with_hooks helper defined here mirrors the dispatch pattern that -AsyncLDClient uses: async hooks are awaited, running in list order for -before_evaluation and reverse order for after_evaluation. - -Each hook receives its own isolated data dict (matching the sync SDK pattern in -client.py:__execute_before_evaluation). A failing hook logs an error and returns -{} rather than aborting the evaluation (matching __try_execute_stage). +Tests for hook and plugin dispatch in AsyncLDClient. """ -import logging +import asyncio +import contextvars +from typing import List import pytest +from ldclient.async_client import AsyncLDClient +from ldclient.async_config import AsyncConfig from ldclient.context import Context from ldclient.evaluation import EvaluationDetail -from ldclient.hook import AsyncHook, EvaluationSeriesContext, Metadata - -log = logging.getLogger('ldclient') - - -async def _try_execute_stage_async(method, hook_name, coro_or_fn): - """Execute a single hook stage the way AsyncLDClient does, catching and logging any exceptions.""" - try: - return await coro_or_fn() - except BaseException as e: - log.error(f"An error occurred in {method} of the hook {hook_name}: #{e}") - return {} - - -async def _evaluate_with_hooks(hooks, series_context, eval_fn): - """Dispatch before/after hooks around an async evaluation function. - - Each hook gets a fresh empty dict for before_evaluation (matching the sync - SDK's __execute_before_evaluation pattern). The per-hook data returned is - stored and passed back to the corresponding hook in after_evaluation. - A hook that raises an exception is caught and logged; it does not abort - the evaluation or affect other hooks. - """ - # before_evaluation: each hook gets its own isolated {} - hook_data = [] - for hook in hooks: - data = await _try_execute_stage_async( - "beforeEvaluation", hook.metadata.name, - lambda h=hook: h.before_evaluation(series_context, {}) - ) - hook_data.append(data) - - detail = await eval_fn() - - # after_evaluation: reversed order, each hook receives its own before data - for hook, data in reversed(list(zip(hooks, hook_data))): - await _try_execute_stage_async( - "afterEvaluation", hook.metadata.name, - lambda h=hook, d=data: h.after_evaluation(series_context, d, detail) - ) - - return detail - - -class RecordingAsyncHook(AsyncHook): - """Async hook that records calls and threads a counter through data.""" - - def __init__(self, name: str): +from ldclient.hook import EvaluationSeriesContext, Hook, Metadata +from ldclient.plugin import ( + AsyncPlugin, + EnvironmentMetadata, + Plugin, + PluginMetadata +) +from ldclient.testing.mock_async_components import ( + MockAsyncFeatureStore, + MockAsyncUpdateProcessor +) +from ldclient.versioned_data_kind import FEATURES + + +def _offline_config(**kwargs) -> AsyncConfig: + return AsyncConfig("test-sdk-key", offline=True, **kwargs) + + +def _make_flag(key: str, value) -> dict: + return { + 'key': key, + 'version': 1, + 'on': False, + 'offVariation': 0, + 'variations': [value], + 'targets': [], + 'contextTargets': [], + 'rules': [], + 'prerequisites': [], + 'salt': '', + } + + +class RecordingHook(Hook): + def __init__(self, name: str, log: List[str]): self._name = name - self.before_calls: list = [] - self.after_calls: list = [] + self._log = log @property def metadata(self) -> Metadata: return Metadata(name=self._name) - async def before_evaluation(self, series_context: EvaluationSeriesContext, data: dict) -> dict: - self.before_calls.append(series_context.key) - return {**data, self._name + '_before': True} + def before_evaluation(self, series_context: EvaluationSeriesContext, data: dict) -> dict: + self._log.append(f"{self._name}:before:{series_context.key}") + return {**data, 'from_before': self._name} - async def after_evaluation(self, series_context: EvaluationSeriesContext, data: dict, - detail: EvaluationDetail) -> dict: - self.after_calls.append(series_context.key) - return {**data, self._name + '_after': True} + def after_evaluation(self, series_context: EvaluationSeriesContext, data: dict, detail: EvaluationDetail) -> dict: + self._log.append(f"{self._name}:after:{series_context.key}:{data.get('from_before')}:{detail.value}") + return data -@pytest.fixture -def series_context(): - return EvaluationSeriesContext( - key='test-flag', - context=Context.create('user-1'), - default_value=False, - method='variation', - ) - - -@pytest.fixture -def mock_detail(): - return EvaluationDetail(True, 0, {'kind': 'OFF'}) +@pytest.mark.asyncio +async def test_hooks_from_config_run_in_order(): + log: List[str] = [] + config = _offline_config(hooks=[RecordingHook('first', log), RecordingHook('second', log)]) + async with AsyncLDClient(config) as client: + await client.variation('flag', Context.create('user'), 'default') -@pytest.fixture -def eval_fn(mock_detail): - async def _fn(): - return mock_detail - return _fn + assert log == [ + 'first:before:flag', + 'second:before:flag', + 'second:after:flag:second:default', + 'first:after:flag:first:default', + ] @pytest.mark.asyncio -async def test_async_hook_before_and_after_awaited(series_context, eval_fn): - hook = RecordingAsyncHook('async1') - await _evaluate_with_hooks([hook], series_context, eval_fn) +async def test_hook_error_does_not_stop_evaluation(): + log: List[str] = [] - assert hook.before_calls == ['test-flag'] - assert hook.after_calls == ['test-flag'] + class FailingHook(Hook): + @property + def metadata(self) -> Metadata: + return Metadata(name='failing') + def before_evaluation(self, series_context, data): + raise RuntimeError('before failed') -@pytest.mark.asyncio -async def test_hooks_called_in_order(series_context, eval_fn): - """before_evaluation runs in list order; after_evaluation runs in reverse.""" - call_log = [] + def after_evaluation(self, series_context, data, detail): + log.append(f"after:{data}") + return data - class OrderAsyncHook(AsyncHook): - def __init__(self, name): - self._name = name + async with AsyncLDClient(_offline_config(hooks=[FailingHook()])) as client: + result = await client.variation('flag', Context.create('user'), 'default') - @property - def metadata(self): - return Metadata(name=self._name) + assert result == 'default' + assert log == ['after:{}'] - async def before_evaluation(self, sc, data): - call_log.append(self._name + '_before') - return data - async def after_evaluation(self, sc, data, detail): - call_log.append(self._name + '_after') - return data +class _YieldingStore(MockAsyncFeatureStore): + """Yields to the event loop on each read, so concurrent evaluations interleave.""" - hooks = [OrderAsyncHook('first'), OrderAsyncHook('second')] - await _evaluate_with_hooks(hooks, series_context, eval_fn) + def __init__(self, seen: List[str], var: contextvars.ContextVar): + super().__init__() + self._seen = seen + self._var = var - # before: list order (first, second) - # after: reversed (second, first) - assert call_log == ['first_before', 'second_before', 'second_after', 'first_after'] + async def get(self, kind, key): + self._seen.append(self._var.get()) + await asyncio.sleep(0) + return await super().get(kind, key) @pytest.mark.asyncio -async def test_hooks_each_called_with_series_context(series_context, eval_fn): - """Each async hook is invoked with the correct series context.""" - hook_a = RecordingAsyncHook('a') - hook_b = RecordingAsyncHook('b') +async def test_context_variable_set_in_before_stage_is_kept_per_evaluation(): + # A tracing hook makes a span current in the before stage and restores the + # previous context in the after stage. Concurrent evaluations must each see + # only their own value, and the restore must not fail. + current = contextvars.ContextVar('current', default='none') + seen_in_store: List[str] = [] + seen_in_after: List[str] = [] + errors: List[BaseException] = [] + + class ContextVarHook(Hook): + @property + def metadata(self) -> Metadata: + return Metadata(name='context-var') + + def before_evaluation(self, series_context, data): + return {'token': current.set(series_context.context.key)} + + def after_evaluation(self, series_context, data, detail): + seen_in_after.append(f"{series_context.context.key}={current.get()}") + try: + current.reset(data['token']) + except BaseException as e: + errors.append(e) + return data - hooks = [hook_a, hook_b] - await _evaluate_with_hooks(hooks, series_context, eval_fn) + store = _YieldingStore(seen_in_store, current) + await store.force_set(FEATURES, _make_flag('flag', 'on')) + store._initialized = True + config = AsyncConfig( + "test-sdk-key", + feature_store=store, + update_processor_class=MockAsyncUpdateProcessor, + send_events=False, + hooks=[ContextVarHook()], + ) - assert hook_a.before_calls == ['test-flag'] - assert hook_b.before_calls == ['test-flag'] + async with AsyncLDClient(config) as client: + keys = ['a', 'b', 'c'] + results = await asyncio.gather(*(client.variation('flag', Context.create(k), 'off') for k in keys)) + + assert results == ['on', 'on', 'on'] + assert errors == [] + assert sorted(seen_in_store) == keys + assert sorted(seen_in_after) == ['a=a', 'b=b', 'c=c'] + assert current.get() == 'none' @pytest.mark.asyncio -async def test_returns_evaluation_detail(series_context, mock_detail, eval_fn): - hook = RecordingAsyncHook('a') - result = await _evaluate_with_hooks([hook], series_context, eval_fn) - assert result is mock_detail +async def test_async_plugin_hooks_run_and_register_is_awaited(): + log: List[str] = [] + registered: List[AsyncLDClient] = [] + class RecordingPlugin(AsyncPlugin): + @property + def metadata(self) -> PluginMetadata: + return PluginMetadata(name='recording-plugin') -@pytest.mark.asyncio -async def test_no_hooks_returns_detail(series_context, mock_detail, eval_fn): - result = await _evaluate_with_hooks([], series_context, eval_fn) - assert result is mock_detail + async def register(self, client, metadata: EnvironmentMetadata) -> None: + await asyncio.sleep(0) + registered.append(client) + def get_hooks(self, metadata: EnvironmentMetadata) -> List[Hook]: + return [RecordingHook('plugin-hook', log)] -@pytest.mark.asyncio -async def test_data_isolated_per_hook(series_context, eval_fn): - """Each hook receives its own fresh {} for before_evaluation (matching sync SDK pattern). + client = AsyncLDClient(_offline_config(plugins=[RecordingPlugin()])) + async with client: + await client.variation('flag', Context.create('user'), 'default') - Data is not threaded across hooks — each hook's before_evaluation starts - with an empty dict and the result is passed only to that hook's own - after_evaluation. This matches client.py:__execute_before_evaluation. - """ - before_data_received = {} + assert registered == [client] + assert log == ['plugin-hook:before:flag', 'plugin-hook:after:flag:plugin-hook:default'] - class CaptureAsyncHook(AsyncHook): - def __init__(self, name): - self._name = name - @property - def metadata(self): - return Metadata(name=self._name) +def test_config_rejects_non_hook(): + with pytest.raises(TypeError): + _offline_config(hooks=[object()]) # type: ignore[list-item] - async def before_evaluation(self, sc, data): - before_data_received[self._name] = dict(data) - return {**data, self._name + '_key': 1} - async def after_evaluation(self, sc, data, detail): - return data +def test_config_rejects_sync_plugin(): + class SyncPlugin(Plugin): + @property + def metadata(self) -> PluginMetadata: + return PluginMetadata(name='sync-plugin') + + def register(self, client, metadata): + pass - await _evaluate_with_hooks([CaptureAsyncHook('one'), CaptureAsyncHook('two')], series_context, eval_fn) + def get_hooks(self, metadata): + return [] - # Each hook should have received a fresh empty dict, not the other hook's output - assert before_data_received['one'] == {} - assert before_data_received['two'] == {} + with pytest.raises(TypeError): + _offline_config(plugins=[SyncPlugin()]) # type: ignore[list-item] From e749a7407fd29a53081be583dee37410e0ad9d13 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Fri, 30 Oct 2026 12:02:20 -0500 Subject: [PATCH 2/3] chore: Ignore wrong hook types and drop the queued contract-test hook Match the sync client: AsyncConfig and add_hook() ignore items of the wrong type. The async contract-test service uses the sync PostingHook. Remove redundant docstring notes. --- contract-tests/async_client_entity.py | 10 ++-------- contract-tests/hook.py | 21 +-------------------- ldclient/async_client.py | 10 +--------- ldclient/async_config.py | 24 +++--------------------- ldclient/hook.py | 6 ------ ldclient/plugin.py | 4 ---- ldclient/testing/test_async_client.py | 9 +++++---- ldclient/testing/test_async_hooks.py | 12 ++++++------ 8 files changed, 18 insertions(+), 78 deletions(-) diff --git a/contract-tests/async_client_entity.py b/contract-tests/async_client_entity.py index 56a845b5..0698639b 100644 --- a/contract-tests/async_client_entity.py +++ b/contract-tests/async_client_entity.py @@ -2,13 +2,12 @@ import json import logging import sys -from concurrent.futures import ThreadPoolExecutor from typing import Any, Callable, Optional import requests from async_big_segment_store_fixture import AsyncBigSegmentStoreFixture from async_flag_change_listener import AsyncListenerRegistry -from hook import QueuedPostingHook +from hook import PostingHook from ldclient import Context from ldclient.async_client import AsyncLDClient @@ -41,7 +40,6 @@ def __init__(self, tag: str, config_params: dict): self.log = logging.getLogger(tag) self._client: Optional[AsyncLDClient] = None self._listeners: Optional[AsyncListenerRegistry] = None - self._hook_executor: Optional[ThreadPoolExecutor] = None self._config_params = config_params self._tag = tag @@ -94,10 +92,8 @@ async def start(self): opts["send_events"] = False if config_params.get("hooks") is not None: - # One worker keeps the posts from all hooks of this client in order. - self._hook_executor = ThreadPoolExecutor(max_workers=1) opts["hooks"] = [ - QueuedPostingHook(h["name"], h["callbackUri"], h.get("data", {}), h.get("errors", {}), self._hook_executor) + PostingHook(h["name"], h["callbackUri"], h.get("data", {}), h.get("errors", {})) for h in config_params["hooks"]["hooks"] ] @@ -278,8 +274,6 @@ async def close(self): await self._listeners.close_all() if self._client is not None: await self._client.close() - if self._hook_executor is not None: - self._hook_executor.shutdown(wait=False) self.log.info('Test ended') diff --git a/contract-tests/hook.py b/contract-tests/hook.py index 55c912d1..b05481d2 100644 --- a/contract-tests/hook.py +++ b/contract-tests/hook.py @@ -1,4 +1,3 @@ -from concurrent.futures import ThreadPoolExecutor from typing import Optional import requests @@ -47,24 +46,6 @@ def __post(self, stage: str, series_context: EvaluationSeriesContext, data: dict 'reason': detail.reason, } - self._send(self.__callback, payload) + requests.post(self.__callback, json=payload) return {**data, **self.__data.get(stage, {})} - - def _send(self, callback: str, payload: dict): - requests.post(callback, json=payload) - - -class QueuedPostingHook(PostingHook): - """PostingHook for the async client. - - The async client runs hook stages on the event loop, so the hook must not - block. The posts go to a single worker thread, which sends them in order. - """ - - def __init__(self, name: str, callback: str, data: dict, errors: dict, executor: ThreadPoolExecutor): - super().__init__(name, callback, data, errors) - self.__executor = executor - - def _send(self, callback: str, payload: dict): - self.__executor.submit(requests.post, callback, json=payload) diff --git a/ldclient/async_client.py b/ldclient/async_client.py index 9c7fdebe..0038d14a 100644 --- a/ldclient/async_client.py +++ b/ldclient/async_client.py @@ -646,15 +646,10 @@ def add_hook(self, hook: Hook): Hooks provide entrypoints which allow for observation of SDK functions. - The hook stages run synchronously on the event loop, between the - start and the end of each evaluation. They must not block. A hook that - must do I/O should put the work on a queue and do it in the background. - :param hook: - :raises TypeError: if ``hook`` is not a :class:`ldclient.hook.Hook` """ if not isinstance(hook, Hook): - raise TypeError(f"AsyncLDClient requires a Hook; got {type(hook).__name__}") + return self.__hooks.append(hook) @@ -675,9 +670,6 @@ async def __evaluate_with_hooks(self, key: str, context: Context, default_value: if not hooks: return await block() - # The stages run in the caller's task, not in a new task or thread. A hook - # can then change context variables in the before stage (for example, to - # make an OpenTelemetry span current) and restore them in the after stage. series_context = EvaluationSeriesContext(key=key, context=context, default_value=default_value, method=method) hook_data = self.__execute_before_evaluation(hooks, series_context) evaluation_result = await block() diff --git a/ldclient/async_config.py b/ldclient/async_config.py index 4681467a..5090d9e3 100644 --- a/ldclient/async_config.py +++ b/ldclient/async_config.py @@ -130,15 +130,6 @@ class AsyncDataSystemConfig: """An optional fallback synchronizer that will read from FDv1""" -def _require_instances(items: Optional[list], kind: type, name: str) -> list: - if not items: - return [] - for item in items: - if not isinstance(item, kind): - raise TypeError(f"AsyncConfig {name} must be {kind.__name__} instances; got {type(item).__name__}") - return list(items) - - class AsyncConfig(DataSourceBuilderConfig, PrivateAttributesConfig): """Advanced configuration options for the async SDK client. @@ -252,11 +243,8 @@ def __init__( :param http: Optional properties for customizing the client's HTTP/HTTPS behavior. See :class:`ldclient.config.HTTPConfig`. :param application: Optional properties for setting application metadata. See :py:attr:`~application` - :param hooks: Hooks provide entrypoints which allow for observation of SDK functions. The hook - stages run synchronously on the event loop and must not block. See :py:attr:`~hooks`. + :param hooks: Hooks provide entrypoints which allow for observation of SDK functions. :param plugins: A list of plugins to be used with the SDK. Plugin support is currently experimental and subject to change. - :raises TypeError: if an item in ``hooks`` is not a :class:`ldclient.hook.Hook`, or an item in - ``plugins`` is not an :class:`ldclient.plugin.AsyncPlugin`. :param enable_event_compression: Whether or not to enable GZIP compression for outgoing events. :param omit_anonymous_contexts: Sets whether anonymous contexts should be omitted from index and identify events. :param payload_filter_key: The payload filter is used to selectively limited the flags and segments delivered in the data source payload. Payload filtering is not supported with the FDv2 data system, so this has no effect on FDv2 requests. @@ -293,8 +281,8 @@ def __init__( self.__http = http self.__big_segments = AsyncBigSegmentsConfig() if not big_segments else big_segments self.__application = validate_application_info(application or {}, log) - self.__hooks = _require_instances(hooks, Hook, 'hooks') - self.__plugins = _require_instances(plugins, AsyncPlugin, 'plugins') + self.__hooks = [hook for hook in hooks if isinstance(hook, Hook)] if hooks else [] + self.__plugins = [plugin for plugin in plugins if isinstance(plugin, AsyncPlugin)] if plugins else [] self.__enable_event_compression = enable_event_compression self.__omit_anonymous_contexts = omit_anonymous_contexts self.__payload_filter_key = payload_filter_key @@ -447,12 +435,6 @@ def hooks(self) -> List[Hook]: Hooks provide entrypoints which allow for observation of SDK functions. - The async client uses the same :class:`ldclient.hook.Hook` type as the - sync client. The hook stages run synchronously on the event loop, - between the start and the end of each evaluation. They must not block. - A hook that must do I/O should put the work on a queue and do it in the - background. - LaunchDarkly provides integration packages, and most applications will not need to implement their own hooks. """ diff --git a/ldclient/hook.py b/ldclient/hook.py index e4ebb371..56352637 100644 --- a/ldclient/hook.py +++ b/ldclient/hook.py @@ -38,12 +38,6 @@ class Hook(ABC): This class includes default implementations for all hook handlers. This allows LaunchDarkly to expand the list of hook handlers without breaking customer integrations. - - Both :class:`ldclient.client.LDClient` and - :class:`ldclient.async_client.AsyncLDClient` use this class. The async - client calls the handlers synchronously on the event loop, so a handler - must not block. A hook that must do I/O should put the work on a queue and - do it in the background. """ @property diff --git a/ldclient/plugin.py b/ldclient/plugin.py index 06986d32..f6c059b5 100644 --- a/ldclient/plugin.py +++ b/ldclient/plugin.py @@ -162,10 +162,6 @@ def get_hooks(self, metadata: EnvironmentMetadata) -> List[Hook]: This method is called before register() to collect all hooks from plugins. The hooks returned will be added to the SDK's hook configuration. - The async client uses the same :class:`ldclient.hook.Hook` type as the - sync client. - - This method is synchronous and must not do I/O. :param metadata: Metadata about the environment in which the SDK is running :return: A list of hooks to be registered with the SDK diff --git a/ldclient/testing/test_async_client.py b/ldclient/testing/test_async_client.py index 02857e9b..876b6b40 100644 --- a/ldclient/testing/test_async_client.py +++ b/ldclient/testing/test_async_client.py @@ -204,11 +204,12 @@ def after_evaluation(self, series_context, data, detail): @pytest.mark.asyncio -async def test_add_hook_rejects_non_hook(): - """add_hook() raises TypeError when given an object that is not a Hook.""" +async def test_add_hook_ignores_non_hook(): + """add_hook() ignores an object that is not a Hook.""" client = AsyncLDClient(_offline_config()) - with pytest.raises(TypeError): - client.add_hook(object()) # type: ignore[arg-type] + client.add_hook(object()) # type: ignore[arg-type] + async with client: + assert await client.variation('flag', Context.create('user'), 'default') == 'default' @pytest.mark.asyncio diff --git a/ldclient/testing/test_async_hooks.py b/ldclient/testing/test_async_hooks.py index 8cbb2dec..be742e9b 100644 --- a/ldclient/testing/test_async_hooks.py +++ b/ldclient/testing/test_async_hooks.py @@ -188,12 +188,13 @@ def get_hooks(self, metadata: EnvironmentMetadata) -> List[Hook]: assert log == ['plugin-hook:before:flag', 'plugin-hook:after:flag:plugin-hook:default'] -def test_config_rejects_non_hook(): - with pytest.raises(TypeError): - _offline_config(hooks=[object()]) # type: ignore[list-item] +def test_config_ignores_non_hook(): + log: List[str] = [] + config = _offline_config(hooks=[object(), RecordingHook('kept', log)]) # type: ignore[list-item] + assert [hook.metadata.name for hook in config.hooks] == ['kept'] -def test_config_rejects_sync_plugin(): +def test_config_ignores_sync_plugin(): class SyncPlugin(Plugin): @property def metadata(self) -> PluginMetadata: @@ -205,5 +206,4 @@ def register(self, client, metadata): def get_hooks(self, metadata): return [] - with pytest.raises(TypeError): - _offline_config(plugins=[SyncPlugin()]) # type: ignore[list-item] + assert _offline_config(plugins=[SyncPlugin()]).plugins == [] # type: ignore[list-item] From 17e83528854e35a940af449b29dd1c28a783d218 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Mon, 2 Nov 2026 08:24:42 -0600 Subject: [PATCH 3/3] fix: Run hook after stages when an async evaluation is cancelled The after stages get an error detail with the default value, and the exception is re-raised. A hook such as the OpenTelemetry tracing hook can then detach its context and end its span. --- ldclient/async_client.py | 10 +++++- ldclient/testing/test_async_hooks.py | 52 ++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 1 deletion(-) diff --git a/ldclient/async_client.py b/ldclient/async_client.py index 0038d14a..08656245 100644 --- a/ldclient/async_client.py +++ b/ldclient/async_client.py @@ -672,7 +672,15 @@ async def __evaluate_with_hooks(self, key: str, context: Context, default_value: series_context = EvaluationSeriesContext(key=key, context=context, default_value=default_value, method=method) hook_data = self.__execute_before_evaluation(hooks, series_context) - evaluation_result = await block() + try: + evaluation_result = await block() + except BaseException: + # The after stages let hooks release what the before stages set up, for + # example an attached tracing context. They do not await, so they do not + # delay a cancellation. + error_detail = EvaluationDetail(default_value, None, error_reason('EXCEPTION')) + self.__execute_after_evaluation(hooks, series_context, hook_data, error_detail) + raise self.__execute_after_evaluation(hooks, series_context, hook_data, evaluation_result.evaluation_detail) return evaluation_result diff --git a/ldclient/testing/test_async_hooks.py b/ldclient/testing/test_async_hooks.py index be742e9b..af65db36 100644 --- a/ldclient/testing/test_async_hooks.py +++ b/ldclient/testing/test_async_hooks.py @@ -207,3 +207,55 @@ def get_hooks(self, metadata): return [] assert _offline_config(plugins=[SyncPlugin()]).plugins == [] # type: ignore[list-item] + + +@pytest.mark.asyncio +async def test_after_stage_runs_when_evaluation_is_cancelled(): + current = contextvars.ContextVar('current', default='none') + after_details: List[EvaluationDetail] = [] + store_entered = asyncio.Event() + + class ContextVarHook(Hook): + @property + def metadata(self) -> Metadata: + return Metadata(name='context-var') + + def before_evaluation(self, series_context, data): + return {'token': current.set('span')} + + def after_evaluation(self, series_context, data, detail): + after_details.append(detail) + current.reset(data['token']) + return data + + class BlockingStore(MockAsyncFeatureStore): + async def get(self, kind, key): + store_entered.set() + await asyncio.Event().wait() + + store = BlockingStore() + store._initialized = True + config = AsyncConfig( + "test-sdk-key", + feature_store=store, + update_processor_class=MockAsyncUpdateProcessor, + send_events=False, + hooks=[ContextVarHook()], + ) + + async with AsyncLDClient(config) as client: + async def evaluate(): + try: + await client.variation('flag', Context.create('user'), 'off') + finally: + assert current.get() == 'none' + + task = asyncio.create_task(evaluate()) + await store_entered.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert len(after_details) == 1 + assert after_details[0].value == 'off' + assert after_details[0].reason == {'kind': 'ERROR', 'errorKind': 'EXCEPTION'}