From a4272d0be53da11bad0778bb68b7da951d7c5d1f Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Mon, 27 Jul 2026 17:22:18 -0500 Subject: [PATCH 1/7] feat: Add async event processor --- ldclient/impl/events/async_event_processor.py | 280 ++++++++++++ ldclient/impl/events/event_processor.py | 104 +---- .../impl/events/event_processor_common.py | 121 ++++- .../impl/events/test_async_event_processor.py | 430 ++++++++++++++++++ 4 files changed, 833 insertions(+), 102 deletions(-) create mode 100644 ldclient/impl/events/async_event_processor.py create mode 100644 ldclient/testing/impl/events/test_async_event_processor.py diff --git a/ldclient/impl/events/async_event_processor.py b/ldclient/impl/events/async_event_processor.py new file mode 100644 index 00000000..e87c0844 --- /dev/null +++ b/ldclient/impl/events/async_event_processor.py @@ -0,0 +1,280 @@ +""" +Implementation details of the analytics event delivery component. +""" + +import asyncio +import gzip +import json +import queue +import time +import uuid +from collections import namedtuple +from random import Random +from typing import Callable, Optional, Union + +from ldclient.async_config import AsyncConfig +from ldclient.impl.aio.concurrency import ( + AsyncEvent, + AsyncLock, + AsyncQueue, + AsyncRepeatingTask, + AsyncTaskRunner, + AsyncWorkerPool +) +from ldclient.impl.aio.transport import AsyncHTTPTransport +from ldclient.impl.events.diagnostics import create_diagnostic_init +from ldclient.impl.events.event_processor_common import ( + EventBuffer, + EventDispatcherBase, + EventOutputFormatter +) +from ldclient.impl.events.types import EventInput +from ldclient.impl.lru_cache import SimpleLRUCache +from ldclient.impl.sampler import Sampler +from ldclient.impl.util import ( + _headers, + check_if_error_is_recoverable_and_log, + log +) +from ldclient.interfaces import EventProcessor + +__MAX_FLUSH_THREADS__ = 5 +__CURRENT_EVENT_SCHEMA__ = 4 + + +EventProcessorMessage = namedtuple('EventProcessorMessage', ['type', 'param']) + + +class EventPayloadSendTask: + def __init__(self, http: AsyncHTTPTransport, config: AsyncConfig, formatter: EventOutputFormatter, payload, response_fn: Callable): + self._http = http + self._config = config + self._formatter = formatter + self._payload = payload + self._response_fn = response_fn + + async def run(self): + try: + output_events = self._formatter.make_output_events(self._payload.events, self._payload.summary) + await self._do_send(output_events) + except Exception: + log.warning('Unhandled exception in event processor. Analytics events were not processed.', exc_info=True) + + async def _do_send(self, output_events): + # noinspection PyBroadException + try: + json_body = json.dumps(output_events, separators=(',', ':')) + log.debug('Sending events payload: ' + json_body) + payload_id = str(uuid.uuid4()) + r = await _post_events_with_retry(self._http, self._config, self._config.events_uri, payload_id, json_body, "%d events" % len(output_events)) + if r: + self._response_fn(r) + return r + except Exception as e: + log.warning('Unhandled exception in event processor. Analytics events were not processed. [%s]', e) + + +class DiagnosticEventSendTask: + def __init__(self, http: AsyncHTTPTransport, config: AsyncConfig, event_body: dict): + self._http = http + self._config = config + self._event_body = event_body + + async def run(self): + # noinspection PyBroadException + try: + json_body = json.dumps(self._event_body) + log.debug('Sending diagnostic event: ' + json_body) + await _post_events_with_retry(self._http, self._config, self._config.events_base_uri + '/diagnostic', None, json_body, "diagnostic event") + except Exception as e: + log.warning('Unhandled exception in event processor. Diagnostic event was not sent. [%s]', e) + + +class EventDispatcher(EventDispatcherBase): + def __init__(self, inbox: AsyncQueue, config: AsyncConfig, http_client, diagnostic_accumulator=None): + self._inbox = inbox + self._config = config + # When no client is injected, the transport creates one targeting the + # events URI and owns it (closing it on shutdown); an injected client + # remains owned by the caller. + self._http = AsyncHTTPTransport(config, client=http_client) + self._disabled = False + self._outbox = EventBuffer(config.events_max_pending) + self._context_keys = SimpleLRUCache(config.context_keys_capacity) + self._formatter = EventOutputFormatter(config) + self._last_known_past_time = 0 + self._deduplicated_contexts = 0 + self._diagnostic_accumulator = None if config.diagnostic_opt_out else diagnostic_accumulator + self._sampler = Sampler(Random()) + self._omit_anonymous_contexts = config.omit_anonymous_contexts + + self._flush_workers = AsyncWorkerPool(__MAX_FLUSH_THREADS__, "ldclient.flush") + self._diagnostic_flush_workers: Optional[AsyncWorkerPool] = None + if self._diagnostic_accumulator is not None: + self._diagnostic_flush_workers = AsyncWorkerPool(1, "ldclient.events.diag_flush") + init_event = create_diagnostic_init(self._diagnostic_accumulator.data_since_date, self._diagnostic_accumulator.diagnostic_id, config) + task = DiagnosticEventSendTask(self._http, self._config, init_event) + self._diagnostic_flush_workers.execute(task.run) + + self._runner = AsyncTaskRunner() + self._runner.spawn("ldclient.events.processor", self._run_main_loop) + + async def _run_main_loop(self): + log.info("Starting event processor") + while True: + try: + message = await self._inbox.get() + if message.type == 'event': + self._process_event(message.param) + elif message.type == 'flush': + self._trigger_flush() + elif message.type == 'flush_contexts': + self._context_keys.clear() + elif message.type == 'diagnostic': + self._send_and_reset_diagnostics() + elif message.type == 'test_sync': + await self._flush_workers.wait() + if self._diagnostic_flush_workers is not None: + await self._diagnostic_flush_workers.wait() + message.param.set() + elif message.type == 'stop': + await self._do_shutdown() + message.param.set() + return + except Exception: + log.error('Unhandled exception in event processor', exc_info=True) + + def _trigger_flush(self): + if self._disabled: + return + payload = self._outbox.get_payload() + if self._diagnostic_accumulator: + self._diagnostic_accumulator.record_events_in_batch(len(payload.events)) + if len(payload.events) > 0 or not payload.summary.is_empty(): + task = EventPayloadSendTask(self._http, self._config, self._formatter, payload, self._handle_response) + if self._flush_workers.execute(task.run): + # The events have been handed off to a flush worker; clear them from our buffer. + self._outbox.clear() + else: + # We're already at our limit of concurrent flushes; leave the events in the buffer. + pass + + def _send_and_reset_diagnostics(self): + if self._diagnostic_accumulator is not None and self._diagnostic_flush_workers is not None: + dropped_event_count = self._outbox.get_and_clear_dropped_count() + stats_event = self._diagnostic_accumulator.create_event_and_reset(dropped_event_count, self._deduplicated_contexts) + self._deduplicated_contexts = 0 + task = DiagnosticEventSendTask(self._http, self._config, stats_event) + self._diagnostic_flush_workers.execute(task.run) + + async def _do_shutdown(self): + self._flush_workers.stop() + await self._flush_workers.wait() + + if self._diagnostic_flush_workers is not None: + self._diagnostic_flush_workers.stop() + await self._diagnostic_flush_workers.wait() + + await self._http.close() + + +class AsyncEventProcessor(EventProcessor): + def __init__(self, config: AsyncConfig, http=None, dispatcher_class=None, diagnostic_accumulator=None): + self._inbox = AsyncQueue(config.events_max_pending) + self._inbox_full = False + self._flush_timer = AsyncRepeatingTask("ldclient.events.flush", config.flush_interval, config.flush_interval, self.flush) + self._contexts_flush_timer = AsyncRepeatingTask("ldclient.events.context-flush", config.context_keys_flush_interval, config.context_keys_flush_interval, self._flush_contexts) + self._flush_timer.start() + self._contexts_flush_timer.start() + self._diagnostic_event_timer: Optional[AsyncRepeatingTask] + if diagnostic_accumulator is not None: + self._diagnostic_event_timer = AsyncRepeatingTask("ldclient.events.send-diagnostic", config.diagnostic_recording_interval, config.diagnostic_recording_interval, self._send_diagnostic) + self._diagnostic_event_timer.start() + else: + self._diagnostic_event_timer = None + + self._close_lock = AsyncLock() + self._closed = False + + (dispatcher_class or EventDispatcher)(self._inbox, config, http, diagnostic_accumulator) + + def send_event(self, event: EventInput): + self._post_to_inbox(EventProcessorMessage('event', event)) + + def flush(self): + self._post_to_inbox(EventProcessorMessage('flush', None)) + + async def stop(self): + async with self._close_lock: + if self._closed: + return + self._closed = True + self._flush_timer.stop() + self._contexts_flush_timer.stop() + if self._diagnostic_event_timer: + self._diagnostic_event_timer.stop() + self.flush() + # Note that here we are not calling _post_to_inbox, because we *do* want to wait if the inbox + # is full; an orderly shutdown can't happen unless these messages are received. + await self._post_message_and_wait('stop') + + def _post_to_inbox(self, message: EventProcessorMessage): + try: + self._inbox.put_nowait(message) + except queue.Full: + if not self._inbox_full: + # possible race condition here, but it's of no real consequence - we'd just get an extra log line + self._inbox_full = True + log.warning("Events are being produced faster than they can be processed; some events will be dropped") + + async def _flush_contexts(self): + await self._inbox.put(EventProcessorMessage('flush_contexts', None)) + + async def _send_diagnostic(self): + await self._inbox.put(EventProcessorMessage('diagnostic', None)) + + # Used only in tests + async def _wait_until_inactive(self): + await self._post_message_and_wait('test_sync') + + async def _post_message_and_wait(self, type): + reply = AsyncEvent() + await self._inbox.put(EventProcessorMessage(type, reply)) + await reply.wait() + + # These magic methods allow use of the "with" block in tests + async def __aenter__(self): + return self + + async def __aexit__(self, type, value, traceback): + await self.stop() + + +async def _post_events_with_retry(http_client: AsyncHTTPTransport, config: AsyncConfig, uri: str, payload_id: Optional[str], body: str, events_description: str): + hdrs = _headers(config) + hdrs['Content-Type'] = 'application/json' + if config.enable_event_compression: + hdrs['Content-Encoding'] = 'gzip' + + if payload_id: + hdrs['X-LaunchDarkly-Event-Schema'] = str(__CURRENT_EVENT_SCHEMA__) + hdrs['X-LaunchDarkly-Payload-ID'] = payload_id + can_retry = True + context = "posting %s" % events_description + data: Union[bytes, str] = gzip.compress(bytes(body, 'utf-8')) if config.enable_event_compression else body + while True: + next_action_message = "will retry" if can_retry else "some events were dropped" + try: + r = await http_client.request('POST', uri, headers=hdrs, body=data) + if r.status < 300: + return r + recoverable = check_if_error_is_recoverable_and_log(context, r.status, None, next_action_message) + if not recoverable: + return r + except Exception as e: + check_if_error_is_recoverable_and_log(context, None, str(e), next_action_message) + if not can_retry: + return None + can_retry = False + # fixed delay of 1 second for event retries + await asyncio.sleep(1) diff --git a/ldclient/impl/events/event_processor.py b/ldclient/impl/events/event_processor.py index 97792fc7..f9ee3ddb 100644 --- a/ldclient/impl/events/event_processor.py +++ b/ldclient/impl/events/event_processor.py @@ -7,30 +7,20 @@ import queue import time import uuid -from calendar import timegm from collections import namedtuple -from email.utils import parsedate from random import Random from threading import Event, Lock, Thread -from typing import Any, Callable, Optional import urllib3 from ldclient.config import Config -from ldclient.context import Context from ldclient.impl.events.diagnostics import create_diagnostic_init from ldclient.impl.events.event_processor_common import ( - DebugEvent, EventBuffer, - EventOutputFormatter, - IndexEvent -) -from ldclient.impl.events.types import ( - EventInput, - EventInputCustom, - EventInputEvaluation, - EventInputIdentify + EventDispatcherBase, + EventOutputFormatter ) +from ldclient.impl.events.types import EventInput from ldclient.impl.fixed_thread_pool import FixedThreadPool from ldclient.impl.http import _http_factory from ldclient.impl.lru_cache import SimpleLRUCache @@ -39,12 +29,9 @@ from ldclient.impl.util import ( _headers, check_if_error_is_recoverable_and_log, - current_time_millis, - is_http_error_recoverable, log ) from ldclient.interfaces import EventProcessor -from ldclient.migrations.tracker import MigrationOpEvent __MAX_FLUSH_THREADS__ = 5 __CURRENT_EVENT_SCHEMA__ = 4 @@ -98,7 +85,7 @@ def run(self): log.warning('Unhandled exception in event processor. Diagnostic event was not sent. [%s]', e) -class EventDispatcher: +class EventDispatcher(EventDispatcherBase): def __init__(self, inbox, config, http_client, diagnostic_accumulator=None): self._inbox = inbox self._config = config @@ -150,78 +137,6 @@ def _run_main_loop(self): except Exception as e: log.error('Unhandled exception in event processor', exc_info=True) - def _process_event(self, event: EventInput): - if self._disabled: - return - - # Decide whether to add the event to the payload. Feature events may be added twice, once for - # the event (if tracked) and once for debugging. - context = None # type: Optional[Context] - full_event = None # type: Any - debug_event = None # type: Optional[DebugEvent] - sampling_ratio = 1 if event.sampling_ratio is None else event.sampling_ratio - - if isinstance(event, EventInputEvaluation): - context = event.context - if not event.exclude_from_summaries: - self._outbox.add_to_summary(event) - if event.track_events: - full_event = event - if self._should_debug_event(event): - debug_event = DebugEvent(event) - elif isinstance(event, EventInputIdentify): - if self._omit_anonymous_contexts: - context = event.context.without_anonymous_contexts() - if not context.valid: - return - - event = EventInputIdentify(event.timestamp, context, event.sampling_ratio) - - full_event = event - elif isinstance(event, EventInputCustom): - context = event.context - full_event = event - elif isinstance(event, MigrationOpEvent): - full_event = event - - self._get_indexable_context(event, lambda c: self._outbox.add_event(IndexEvent(event.timestamp, c))) - - if full_event and self._sampler.sample(sampling_ratio): - self._outbox.add_event(full_event) - - if debug_event and self._sampler.sample(sampling_ratio): - self._outbox.add_event(debug_event) - - def _get_indexable_context(self, event: EventInput, block: Callable[[Context], None]): - if event.context is None: - return - - context = event.context - if self._omit_anonymous_contexts: - context = context.without_anonymous_contexts() - - if not context.valid: - return - - already_seen = self._context_keys.put(context.fully_qualified_key, True) - if already_seen: - self._deduplicated_contexts += 1 - return - elif isinstance(event, EventInputIdentify) or isinstance(event, MigrationOpEvent): - return - - block(context) - - def _should_debug_event(self, event: EventInputEvaluation): - if event.flag is None: - return False - debug_until = event.flag.debug_events_until_date - if debug_until is not None: - last_past = self._last_known_past_time - if debug_until > last_past and debug_until > current_time_millis(): - return True - return False - def _trigger_flush(self): if self._disabled: return @@ -237,17 +152,6 @@ def _trigger_flush(self): # We're already at our limit of concurrent flushes; leave the events in the buffer. pass - def _handle_response(self, r): - server_date_str = r.headers.get('Date') - if server_date_str is not None: - server_date = parsedate(server_date_str) - if server_date is not None: - timestamp = int(timegm(server_date) * 1000) - self._last_known_past_time = timestamp - if r.status > 299 and not is_http_error_recoverable(r.status): - self._disabled = True - return - def _send_and_reset_diagnostics(self): if self._diagnostic_accumulator is not None: dropped_event_count = self._outbox.get_and_clear_dropped_count() diff --git a/ldclient/impl/events/event_processor_common.py b/ldclient/impl/events/event_processor_common.py index 1b558aed..b8b60827 100644 --- a/ldclient/impl/events/event_processor_common.py +++ b/ldclient/impl/events/event_processor_common.py @@ -5,19 +5,29 @@ and async (async_event_processor.py) implementations. """ +from calendar import timegm from collections import namedtuple -from typing import Any, Dict, List +from email.utils import parsedate +from typing import Any, Callable, Dict, List, Optional from ldclient.config import PrivateAttributesConfig from ldclient.context import Context from ldclient.impl.events.event_context_formatter import EventContextFormatter from ldclient.impl.events.event_summarizer import EventSummarizer, EventSummary from ldclient.impl.events.types import ( + EventInput, EventInputCustom, EventInputEvaluation, EventInputIdentify ) -from ldclient.impl.util import log, timedelta_millis +from ldclient.impl.lru_cache import SimpleLRUCache +from ldclient.impl.sampler import Sampler +from ldclient.impl.util import ( + current_time_millis, + is_http_error_recoverable, + log, + timedelta_millis +) from ldclient.migrations.tracker import MigrationOpEvent # --------------------------------------------------------------------------- @@ -224,3 +234,110 @@ def _base_eval_props(self, e: EventInputEvaluation, kind: str) -> dict: if e.prereq_of is not None: out['prereqOf'] = e.prereq_of.key return out + + +# --------------------------------------------------------------------------- +# EventDispatcherBase — pure event-handling logic shared by both dispatchers +# --------------------------------------------------------------------------- + +class EventDispatcherBase: + """ + Pure event-handling methods shared by the sync and async EventDispatcher + implementations. These methods perform no I/O. + + This class does not define ``__init__``. Subclasses are responsible for + setting the following attributes, which these methods rely on: + ``_disabled``, ``_outbox`` (an :class:`EventBuffer`), ``_sampler``, + ``_context_keys``, ``_last_known_past_time``, ``_omit_anonymous_contexts``, + and ``_deduplicated_contexts``. + """ + + _disabled: bool + _outbox: EventBuffer + _sampler: Sampler + _context_keys: SimpleLRUCache + _last_known_past_time: int + _omit_anonymous_contexts: bool + _deduplicated_contexts: int + + def _process_event(self, event: EventInput): + if self._disabled: + return + + # Decide whether to add the event to the payload. Feature events may be added twice, once for + # the event (if tracked) and once for debugging. + context: Optional[Context] = None + full_event: Any = None + debug_event: Optional[DebugEvent] = None + sampling_ratio = 1 if event.sampling_ratio is None else event.sampling_ratio + + if isinstance(event, EventInputEvaluation): + context = event.context + if not event.exclude_from_summaries: + self._outbox.add_to_summary(event) + if event.track_events: + full_event = event + if self._should_debug_event(event): + debug_event = DebugEvent(event) + elif isinstance(event, EventInputIdentify): + if self._omit_anonymous_contexts: + context = event.context.without_anonymous_contexts() + if not context.valid: + return + + event = EventInputIdentify(event.timestamp, context, event.sampling_ratio) + + full_event = event + elif isinstance(event, EventInputCustom): + context = event.context + full_event = event + elif isinstance(event, MigrationOpEvent): + full_event = event + + self._get_indexable_context(event, lambda c: self._outbox.add_event(IndexEvent(event.timestamp, c))) + + if full_event and self._sampler.sample(sampling_ratio): + self._outbox.add_event(full_event) + + if debug_event and self._sampler.sample(sampling_ratio): + self._outbox.add_event(debug_event) + + def _get_indexable_context(self, event: EventInput, block: Callable[[Context], None]): + if event.context is None: + return + + context = event.context + if self._omit_anonymous_contexts: + context = context.without_anonymous_contexts() + + if not context.valid: + return + + already_seen = self._context_keys.put(context.fully_qualified_key, True) + if already_seen: + self._deduplicated_contexts += 1 + return + elif isinstance(event, EventInputIdentify) or isinstance(event, MigrationOpEvent): + return + + block(context) + + def _should_debug_event(self, event: EventInputEvaluation): + if event.flag is None: + return False + debug_until = event.flag.debug_events_until_date + if debug_until is not None: + last_past = self._last_known_past_time + if debug_until > last_past and debug_until > current_time_millis(): + return True + return False + + def _handle_response(self, r): + server_date_str = r.headers.get('Date') + if server_date_str is not None: + server_date = parsedate(server_date_str) + if server_date is not None: + timestamp = int(timegm(server_date) * 1000) + self._last_known_past_time = timestamp + if r.status > 299 and not is_http_error_recoverable(r.status): + self._disabled = True diff --git a/ldclient/testing/impl/events/test_async_event_processor.py b/ldclient/testing/impl/events/test_async_event_processor.py new file mode 100644 index 00000000..dac75bf8 --- /dev/null +++ b/ldclient/testing/impl/events/test_async_event_processor.py @@ -0,0 +1,430 @@ +""" +Tests for AsyncEventProcessor. + +These mirror the sync DefaultEventProcessor tests: the processor is driven +through its public API plus the ``_wait_until_inactive`` test handshake, and +HTTP delivery is captured by a mock aiohttp-style session injected through +the ``http`` constructor parameter. +""" + +import asyncio +import gzip +import json +import uuid +from contextlib import asynccontextmanager +from email.utils import formatdate +from typing import Any, List, Optional, Tuple + +import pytest + +from ldclient.async_config import AsyncConfig +from ldclient.context import Context +from ldclient.impl.events.async_event_processor import AsyncEventProcessor +from ldclient.impl.events.diagnostics import ( + _DiagnosticAccumulator, + create_diagnostic_id +) +from ldclient.impl.events.types import ( + EventInputCustom, + EventInputEvaluation, + EventInputIdentify +) +from ldclient.testing.builders import FlagBuilder +from ldclient.testing.stub_util import _MockHTTPHeaderDict + +pytestmark = pytest.mark.asyncio + +context = Context.builder('userkey').name('Red').build() +flag = FlagBuilder('flagkey').version(2).build() +timestamp = 10000 + + +# --------------------------------------------------------------------------- +# Mock aiohttp session (mirrors stub_util.MockHttp for the aiohttp surface +# used by AsyncHTTPTransport) +# --------------------------------------------------------------------------- + +class MockAioResponse: + def __init__(self, status: int, headers: dict): + self.status = status + self.headers = _MockHTTPHeaderDict(headers) + + async def text(self, encoding: str = 'UTF-8', errors: str = 'strict') -> str: + return '' + + +class _MockRequestContext: + def __init__(self, response: MockAioResponse): + self._response = response + + async def __aenter__(self) -> MockAioResponse: + return self._response + + async def __aexit__(self, exc_type, exc_value, traceback) -> bool: + return False + + +class MockAioHttp: + def __init__(self): + self._recorded_requests: List[Tuple[Any, Any]] = [] + self._response_status = 200 + self._server_time: Optional[int] = None + + def request(self, method, uri, headers=None, data=None, timeout=None, proxy=None): + self._recorded_requests.append((headers, data)) + resp_hdr = dict() + if self._server_time is not None: + resp_hdr['date'] = formatdate(self._server_time / 1000, localtime=False, usegmt=True) + return _MockRequestContext(MockAioResponse(self._response_status, resp_hdr)) + + @property + def request_data(self): + if len(self._recorded_requests) != 0: + return self._recorded_requests[-1][1] + + @property + def request_headers(self): + if len(self._recorded_requests) != 0: + return self._recorded_requests[-1][0] + + @property + def recorded_requests(self): + return self._recorded_requests + + def set_response_status(self, status: int): + self._response_status = status + + def set_server_time(self, timestamp: int): + self._server_time = timestamp + + def reset(self): + self._recorded_requests = [] + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +@asynccontextmanager +async def make_processor(mock_http: MockAioHttp, **kwargs): + kwargs.setdefault('diagnostic_opt_out', True) + kwargs.setdefault('sdk_key', 'SDK_KEY') + config = AsyncConfig(**kwargs) + diagnostic_accumulator = _DiagnosticAccumulator(create_diagnostic_id(config)) + ep = AsyncEventProcessor(config, mock_http, diagnostic_accumulator=diagnostic_accumulator) + try: + yield ep + finally: + await ep.stop() + + +async def flush_and_get_events(ep: AsyncEventProcessor, mock_http: MockAioHttp): + ep.flush() + await ep._wait_until_inactive() + if mock_http.request_data is None: + raise AssertionError('Expected to get an HTTP request but did not get one') + return json.loads(mock_http.request_data) + + +# --------------------------------------------------------------------------- +# Event payload tests +# --------------------------------------------------------------------------- + +async def test_identify_event_is_queued(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + e = EventInputIdentify(timestamp, context) + ep.send_event(e) + + output = await flush_and_get_events(ep, mock_http) + assert len(output) == 1 + assert output[0] == {'kind': 'identify', 'creationDate': timestamp, 'context': context.to_dict()} + + +async def test_individual_feature_event_is_queued_with_index_event(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + e = EventInputEvaluation(timestamp, context, flag.key, flag, 1, 'value', None, 'default', None, True) + ep.send_event(e) + + output = await flush_and_get_events(ep, mock_http) + assert len(output) == 3 + assert output[0]['kind'] == 'index' + assert output[0]['context'] == context.to_dict() + assert output[1]['kind'] == 'feature' + assert output[1]['key'] == flag.key + assert output[1]['value'] == 'value' + assert output[2]['kind'] == 'summary' + + +async def test_custom_event_is_queued(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + e = EventInputCustom(timestamp, context, 'eventkey', {'thing': 'stuff'}, 1.5) + ep.send_event(e) + + output = await flush_and_get_events(ep, mock_http) + assert len(output) == 2 + assert output[0]['kind'] == 'index' + assert output[1]['kind'] == 'custom' + assert output[1]['key'] == 'eventkey' + assert output[1]['data'] == {'thing': 'stuff'} + assert output[1]['metricValue'] == 1.5 + + +async def test_two_events_for_same_context_only_produce_one_index_event(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + e0 = EventInputEvaluation(timestamp, context, flag.key, flag, 1, 'value1', None, 'default', None, True) + e1 = EventInputEvaluation(timestamp, context, flag.key, flag, 2, 'value2', None, 'default', None, True) + ep.send_event(e0) + ep.send_event(e1) + + output = await flush_and_get_events(ep, mock_http) + assert len(output) == 4 + assert output[0]['kind'] == 'index' + assert output[1]['kind'] == 'feature' + assert output[2]['kind'] == 'feature' + assert output[3]['kind'] == 'summary' + + +async def test_nontracked_events_are_summarized(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + e = EventInputEvaluation(timestamp, context, flag.key, flag, 1, 'value', None, 'default', None, False) + ep.send_event(e) + + output = await flush_and_get_events(ep, mock_http) + assert len(output) == 2 + assert output[0]['kind'] == 'index' + assert output[1]['kind'] == 'summary' + assert flag.key in output[1]['features'] + + +async def test_nothing_is_sent_if_there_are_no_events(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + ep.flush() + await ep._wait_until_inactive() + assert mock_http.request_data is None + + +async def test_stop_flushes_remaining_events(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + # stop() (via the context manager) triggers a final flush + assert mock_http.request_data is not None + output = json.loads(mock_http.request_data) + assert output[0]['kind'] == 'identify' + + +# --------------------------------------------------------------------------- +# Header tests +# --------------------------------------------------------------------------- + +async def test_sdk_key_is_sent(): + mock_http = MockAioHttp() + async with make_processor(mock_http, sdk_key='SDK_KEY') as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + await flush_and_get_events(ep, mock_http) + + assert mock_http.request_headers.get('Authorization') == 'SDK_KEY' + + +async def test_event_schema_set_on_event_send(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + await flush_and_get_events(ep, mock_http) + + assert mock_http.request_headers.get('X-LaunchDarkly-Event-Schema') == "4" + + +async def test_event_payload_id_is_sent(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + await flush_and_get_events(ep, mock_http) + + header_val = mock_http.request_headers.get('X-LaunchDarkly-Payload-ID') + assert header_val is not None + # Throws on invalid UUID + uuid.UUID(header_val) + + +async def test_event_payload_id_changes_between_requests(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + await flush_and_get_events(ep, mock_http) + + ep.send_event(EventInputIdentify(timestamp, context)) + ep.flush() + await ep._wait_until_inactive() + + first_payload_id = mock_http.recorded_requests[0][0].get('X-LaunchDarkly-Payload-ID') + second_payload_id = mock_http.recorded_requests[1][0].get('X-LaunchDarkly-Payload-ID') + assert first_payload_id != second_payload_id + + +# --------------------------------------------------------------------------- +# Diagnostics +# --------------------------------------------------------------------------- + +async def test_init_diagnostic_event_sent(): + mock_http = MockAioHttp() + async with make_processor(mock_http, diagnostic_opt_out=False) as ep: + diag_init = await flush_and_get_events(ep, mock_http) + assert len(diag_init) == 6 + assert diag_init['kind'] == 'diagnostic-init' + + +async def test_periodic_diagnostic_includes_events_in_batch(): + mock_http = MockAioHttp() + async with make_processor(mock_http, diagnostic_opt_out=False) as ep: + # Ignore init event + await flush_and_get_events(ep, mock_http) + # Send a payload with a single event + ep.send_event(EventInputIdentify(timestamp, context)) + await flush_and_get_events(ep, mock_http) + + await ep._send_diagnostic() + diag_event = await flush_and_get_events(ep, mock_http) + assert len(diag_event) == 8 + assert diag_event['kind'] == 'diagnostic' + assert diag_event['eventsInLastBatch'] == 1 + assert diag_event['deduplicatedUsers'] == 0 + + +async def test_periodic_diagnostic_includes_deduplicated_users(): + mock_http = MockAioHttp() + async with make_processor(mock_http, diagnostic_opt_out=False) as ep: + # Ignore init event + await flush_and_get_events(ep, mock_http) + # Send two custom events with the same user to cause a user deduplication + e0 = EventInputCustom(timestamp, context, 'event1', None, None) + e1 = EventInputCustom(timestamp, context, 'event2', None, None) + ep.send_event(e0) + ep.send_event(e1) + await flush_and_get_events(ep, mock_http) + + await ep._send_diagnostic() + diag_event = await flush_and_get_events(ep, mock_http) + assert len(diag_event) == 8 + assert diag_event['kind'] == 'diagnostic' + assert diag_event['eventsInLastBatch'] == 3 + assert diag_event['deduplicatedUsers'] == 1 + + +async def test_event_schema_not_set_on_diagnostic_send(): + mock_http = MockAioHttp() + async with make_processor(mock_http, diagnostic_opt_out=False) as ep: + await ep._wait_until_inactive() + assert mock_http.request_headers.get('X-LaunchDarkly-Event-Schema') is None + + +# --------------------------------------------------------------------------- +# HTTP error handling +# --------------------------------------------------------------------------- + +async def verify_unrecoverable_http_error(status: int): + mock_http = MockAioHttp() + async with make_processor(mock_http, sdk_key='SDK_KEY') as ep: + mock_http.set_response_status(status) + ep.send_event(EventInputIdentify(timestamp, context)) + ep.flush() + await ep._wait_until_inactive() + mock_http.reset() + + ep.send_event(EventInputIdentify(timestamp, context)) + ep.flush() + await ep._wait_until_inactive() + assert mock_http.request_data is None + + +async def verify_recoverable_http_error(status: int): + mock_http = MockAioHttp() + async with make_processor(mock_http, sdk_key='SDK_KEY') as ep: + mock_http.set_response_status(status) + ep.send_event(EventInputIdentify(timestamp, context)) + ep.flush() + await ep._wait_until_inactive() + mock_http.reset() + + ep.send_event(EventInputIdentify(timestamp, context)) + ep.flush() + await ep._wait_until_inactive() + assert mock_http.request_data is not None + + +async def test_no_more_payloads_are_sent_after_401_error(): + await verify_unrecoverable_http_error(401) + + +async def test_no_more_payloads_are_sent_after_403_error(): + await verify_unrecoverable_http_error(403) + + +async def test_will_still_send_after_408_error(): + await verify_recoverable_http_error(408) + + +async def test_will_still_send_after_429_error(): + await verify_recoverable_http_error(429) + + +async def test_will_still_send_after_500_error(): + await verify_recoverable_http_error(500) + + +# --------------------------------------------------------------------------- +# Inbox capacity +# --------------------------------------------------------------------------- + +async def test_does_not_block_on_full_inbox(): + config = AsyncConfig("fake_sdk_key", events_max_pending=1) # this sets the size of both the inbox and the outbox to 1 + ep_inbox_holder = [None] + + def dispatcher_factory(inbox, config, http, diag): + ep_inbox_holder[0] = inbox # it's an array because otherwise it's hard for a closure to modify a variable + return None # the dispatcher object itself doesn't matter, we only manipulate the inbox + + async def event_consumer(): + while True: + message = await ep_inbox.get() + if message.type == 'stop': + message.param.set() + return + + mock_http = MockAioHttp() + ep = AsyncEventProcessor(config, mock_http, dispatcher_factory) + ep_inbox = ep_inbox_holder[0] + event1 = EventInputCustom(timestamp, context, 'event1') + event2 = EventInputCustom(timestamp, context, 'event2') + ep.send_event(event1) + ep.send_event(event2) # this event should be dropped - inbox is full + message1 = await ep_inbox.get(block=False) + had_no_more = ep_inbox.empty() + consumer = asyncio.ensure_future(event_consumer()) + await ep.stop() + await consumer + assert message1.param == event1 + assert had_no_more + + +# --------------------------------------------------------------------------- +# Compression +# --------------------------------------------------------------------------- + +async def test_event_payload_is_gzip_compressed_when_enabled(): + mock_http = MockAioHttp() + async with make_processor(mock_http, enable_event_compression=True) as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + ep.flush() + await ep._wait_until_inactive() + + assert mock_http.request_headers.get('Content-Encoding') == 'gzip' + output = json.loads(gzip.decompress(mock_http.request_data)) + assert len(output) == 1 + assert output[0]['kind'] == 'identify' From dfe1754be0ef859884171007f835a38e915fb2bb Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Tue, 28 Jul 2026 09:08:03 -0500 Subject: [PATCH 2/7] refactor: Rename concrete event processor to DefaultAsyncEventProcessor Implement the new AsyncEventProcessor interface with a concrete DefaultAsyncEventProcessor, matching the sync DefaultEventProcessor naming convention. --- ldclient/impl/events/async_event_processor.py | 4 ++-- .../impl/events/test_async_event_processor.py | 12 +++++++----- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/ldclient/impl/events/async_event_processor.py b/ldclient/impl/events/async_event_processor.py index e87c0844..5f236808 100644 --- a/ldclient/impl/events/async_event_processor.py +++ b/ldclient/impl/events/async_event_processor.py @@ -36,7 +36,7 @@ check_if_error_is_recoverable_and_log, log ) -from ldclient.interfaces import EventProcessor +from ldclient.interfaces import AsyncEventProcessor __MAX_FLUSH_THREADS__ = 5 __CURRENT_EVENT_SCHEMA__ = 4 @@ -178,7 +178,7 @@ async def _do_shutdown(self): await self._http.close() -class AsyncEventProcessor(EventProcessor): +class DefaultAsyncEventProcessor(AsyncEventProcessor): def __init__(self, config: AsyncConfig, http=None, dispatcher_class=None, diagnostic_accumulator=None): self._inbox = AsyncQueue(config.events_max_pending) self._inbox_full = False diff --git a/ldclient/testing/impl/events/test_async_event_processor.py b/ldclient/testing/impl/events/test_async_event_processor.py index dac75bf8..4e703651 100644 --- a/ldclient/testing/impl/events/test_async_event_processor.py +++ b/ldclient/testing/impl/events/test_async_event_processor.py @@ -1,5 +1,5 @@ """ -Tests for AsyncEventProcessor. +Tests for DefaultAsyncEventProcessor. These mirror the sync DefaultEventProcessor tests: the processor is driven through its public API plus the ``_wait_until_inactive`` test handshake, and @@ -19,7 +19,9 @@ from ldclient.async_config import AsyncConfig from ldclient.context import Context -from ldclient.impl.events.async_event_processor import AsyncEventProcessor +from ldclient.impl.events.async_event_processor import ( + DefaultAsyncEventProcessor +) from ldclient.impl.events.diagnostics import ( _DiagnosticAccumulator, create_diagnostic_id @@ -111,14 +113,14 @@ async def make_processor(mock_http: MockAioHttp, **kwargs): kwargs.setdefault('sdk_key', 'SDK_KEY') config = AsyncConfig(**kwargs) diagnostic_accumulator = _DiagnosticAccumulator(create_diagnostic_id(config)) - ep = AsyncEventProcessor(config, mock_http, diagnostic_accumulator=diagnostic_accumulator) + ep = DefaultAsyncEventProcessor(config, mock_http, diagnostic_accumulator=diagnostic_accumulator) try: yield ep finally: await ep.stop() -async def flush_and_get_events(ep: AsyncEventProcessor, mock_http: MockAioHttp): +async def flush_and_get_events(ep: DefaultAsyncEventProcessor, mock_http: MockAioHttp): ep.flush() await ep._wait_until_inactive() if mock_http.request_data is None: @@ -398,7 +400,7 @@ async def event_consumer(): return mock_http = MockAioHttp() - ep = AsyncEventProcessor(config, mock_http, dispatcher_factory) + ep = DefaultAsyncEventProcessor(config, mock_http, dispatcher_factory) ep_inbox = ep_inbox_holder[0] event1 = EventInputCustom(timestamp, context, 'event1') event2 = EventInputCustom(timestamp, context, 'event2') From ccb9c58ea2e620d1a89f995cc441682f32841853 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Wed, 29 Jul 2026 13:38:26 -0500 Subject: [PATCH 3/7] feat: Add flush_and_wait to DefaultAsyncEventProcessor Triggers a flush and awaits delivery via a new inbox message, returning whether it completed within the timeout. --- ldclient/impl/events/async_event_processor.py | 11 +++++++ .../impl/events/test_async_event_processor.py | 29 +++++++++++++++++++ 2 files changed, 40 insertions(+) diff --git a/ldclient/impl/events/async_event_processor.py b/ldclient/impl/events/async_event_processor.py index 5f236808..e7d178b8 100644 --- a/ldclient/impl/events/async_event_processor.py +++ b/ldclient/impl/events/async_event_processor.py @@ -132,6 +132,10 @@ async def _run_main_loop(self): self._context_keys.clear() elif message.type == 'diagnostic': self._send_and_reset_diagnostics() + elif message.type == 'flush_and_wait': + self._trigger_flush() + await self._flush_workers.wait() + message.param.set() elif message.type == 'test_sync': await self._flush_workers.wait() if self._diagnostic_flush_workers is not None: @@ -204,6 +208,13 @@ def send_event(self, event: EventInput): def flush(self): self._post_to_inbox(EventProcessorMessage('flush', None)) + async def flush_and_wait(self, timeout: float) -> bool: + try: + await asyncio.wait_for(self._post_message_and_wait('flush_and_wait'), timeout) + return True + except asyncio.TimeoutError: + return False + async def stop(self): async with self._close_lock: if self._closed: diff --git a/ldclient/testing/impl/events/test_async_event_processor.py b/ldclient/testing/impl/events/test_async_event_processor.py index 4e703651..780513c7 100644 --- a/ldclient/testing/impl/events/test_async_event_processor.py +++ b/ldclient/testing/impl/events/test_async_event_processor.py @@ -174,6 +174,35 @@ async def test_custom_event_is_queued(): assert output[1]['metricValue'] == 1.5 +# --------------------------------------------------------------------------- +# flush_and_wait +# --------------------------------------------------------------------------- + +async def test_flush_and_wait_delivers_events_and_returns_true(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + + delivered = await ep.flush_and_wait(5) + + assert delivered is True + assert mock_http.request_data is not None + output = json.loads(mock_http.request_data) + assert len(output) == 1 + assert output[0]['kind'] == 'identify' + + +async def test_flush_and_wait_returns_false_on_timeout(): + mock_http = MockAioHttp() + async with make_processor(mock_http) as ep: + ep.send_event(EventInputIdentify(timestamp, context)) + + # A zero timeout can't complete the flush round-trip, so it reports False. + delivered = await ep.flush_and_wait(0) + + assert delivered is False + + async def test_two_events_for_same_context_only_produce_one_index_event(): mock_http = MockAioHttp() async with make_processor(mock_http) as ep: From 5ed9484adccdba75891acfa58eac1bd92c33a3d8 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 30 Jul 2026 11:18:08 -0500 Subject: [PATCH 4/7] refactor: Replace AsyncWorkerPool with BoundedTaskSet The async event delivery concurrency limiter no longer mirrors the sync FixedThreadPool's shape. BoundedTaskSet drops the unused name parameter and the thread vocabulary, reserves a slot synchronously at spawn (so a full set rejects rather than queues), and uses a done-callback plus asyncio.gather for cleanup and draining. --- ldclient/impl/aio/concurrency.py | 71 ++++++++----------- ldclient/impl/events/async_event_processor.py | 26 +++---- ldclient/testing/test_aio.py | 37 +++++----- 3 files changed, 63 insertions(+), 71 deletions(-) diff --git a/ldclient/impl/aio/concurrency.py b/ldclient/impl/aio/concurrency.py index fdc51403..beafd56b 100644 --- a/ldclient/impl/aio/concurrency.py +++ b/ldclient/impl/aio/concurrency.py @@ -2,7 +2,7 @@ Async concurrency primitives used by the async data source, event processor, and data system. Each wraps a piece of fiddly asyncio plumbing (timeout-aware waits, queue exception normalization, an interval-from-start repeating task, a bounded -task pool) that callers would otherwise inline repeatedly. The sync code uses the +task set) that callers would otherwise inline repeatedly. The sync code uses the equivalent stdlib/SDK primitives (``threading.Event``/``Lock``, ``queue.Queue``, ``RepeatingTask``, ``FixedThreadPool``) directly, so these have no sync twin. """ @@ -12,7 +12,7 @@ import time from queue import Empty as QueueEmpty # noqa: F401 (shared timeout exception) from queue import Full as QueueFull # noqa: F401 (shared capacity exception) -from typing import Any, Callable, Optional, Set +from typing import Any, Callable, Coroutine, Optional, Set from ldclient.impl.util import log @@ -250,50 +250,37 @@ async def _run(self): pass -class AsyncWorkerPool: - """A fixed-size pool of concurrent tasks that rejects jobs when its limit - is reached. Matches the contract of - ``ldclient.impl.fixed_thread_pool.FixedThreadPool``.""" +class BoundedTaskSet: + """Runs up to ``limit`` coroutines concurrently as background tasks. When the + limit is reached, ``try_run`` rejects new work (returning False) rather than + queuing it, so callers can apply their own backpressure. ``drain`` awaits all + in-flight tasks; ``stop`` prevents any further work from being accepted.""" - def __init__(self, size: int, name: str): - self._size = size - self._name = name - self._busy: Set[asyncio.Task] = set() - self._event = AsyncEvent() - self._stopped = False + def __init__(self, limit: int): + self._limit = limit + self._tasks: Set[asyncio.Task] = set() + self._accepting = True - def execute(self, jobFn: Callable) -> bool: - """Schedules a job for execution if the pool is not already at its - limit, and returns True if successful; returns False if all workers - are busy.""" - if self._stopped or len(self._busy) >= self._size: + def try_run(self, job: Callable[[], Coroutine[Any, Any, Any]]) -> bool: + """Runs ``job()`` as a background task if fewer than ``limit`` tasks are + already running and the set is still accepting work. Returns True if the + task was started, or False if the set is full or has been stopped.""" + if not self._accepting or len(self._tasks) >= self._limit: return False - task = asyncio.ensure_future(self._run_job(jobFn)) - self._busy.add(task) + task = asyncio.create_task(job()) + self._tasks.add(task) + task.add_done_callback(self._on_done) return True - async def _run_job(self, jobFn: Callable) -> None: - try: - result = jobFn() - if inspect.isawaitable(result): - await result - except Exception: - log.warning('Unhandled exception in worker thread', exc_info=True) - finally: - task = asyncio.current_task() - if task is not None: - self._busy.discard(task) - self._event.set() - - async def wait(self) -> None: - """Waits until all currently busy workers have completed their jobs.""" - while len(self._busy) > 0: - self._event.clear() - if len(self._busy) == 0: - return - await self._event.wait() + def _on_done(self, task: asyncio.Task) -> None: + self._tasks.discard(task) + if not task.cancelled() and task.exception() is not None: + log.warning('Unhandled exception in background task', exc_info=task.exception()) + + async def drain(self) -> None: + """Waits for all currently running tasks to complete.""" + await asyncio.gather(*self._tasks, return_exceptions=True) def stop(self) -> None: - """Tells the pool to reject any further jobs; active jobs run to - completion.""" - self._stopped = True + """Rejects any further work; tasks already running finish normally.""" + self._accepting = False diff --git a/ldclient/impl/events/async_event_processor.py b/ldclient/impl/events/async_event_processor.py index e7d178b8..ac9b2a5a 100644 --- a/ldclient/impl/events/async_event_processor.py +++ b/ldclient/impl/events/async_event_processor.py @@ -19,7 +19,7 @@ AsyncQueue, AsyncRepeatingTask, AsyncTaskRunner, - AsyncWorkerPool + BoundedTaskSet ) from ldclient.impl.aio.transport import AsyncHTTPTransport from ldclient.impl.events.diagnostics import create_diagnostic_init @@ -38,7 +38,7 @@ ) from ldclient.interfaces import AsyncEventProcessor -__MAX_FLUSH_THREADS__ = 5 +__MAX_FLUSH_CONCURRENCY__ = 5 __CURRENT_EVENT_SCHEMA__ = 4 @@ -108,13 +108,13 @@ def __init__(self, inbox: AsyncQueue, config: AsyncConfig, http_client, diagnost self._sampler = Sampler(Random()) self._omit_anonymous_contexts = config.omit_anonymous_contexts - self._flush_workers = AsyncWorkerPool(__MAX_FLUSH_THREADS__, "ldclient.flush") - self._diagnostic_flush_workers: Optional[AsyncWorkerPool] = None + self._flush_workers = BoundedTaskSet(__MAX_FLUSH_CONCURRENCY__) + self._diagnostic_flush_workers: Optional[BoundedTaskSet] = None if self._diagnostic_accumulator is not None: - self._diagnostic_flush_workers = AsyncWorkerPool(1, "ldclient.events.diag_flush") + self._diagnostic_flush_workers = BoundedTaskSet(1) init_event = create_diagnostic_init(self._diagnostic_accumulator.data_since_date, self._diagnostic_accumulator.diagnostic_id, config) task = DiagnosticEventSendTask(self._http, self._config, init_event) - self._diagnostic_flush_workers.execute(task.run) + self._diagnostic_flush_workers.try_run(task.run) self._runner = AsyncTaskRunner() self._runner.spawn("ldclient.events.processor", self._run_main_loop) @@ -134,12 +134,12 @@ async def _run_main_loop(self): self._send_and_reset_diagnostics() elif message.type == 'flush_and_wait': self._trigger_flush() - await self._flush_workers.wait() + await self._flush_workers.drain() message.param.set() elif message.type == 'test_sync': - await self._flush_workers.wait() + await self._flush_workers.drain() if self._diagnostic_flush_workers is not None: - await self._diagnostic_flush_workers.wait() + await self._diagnostic_flush_workers.drain() message.param.set() elif message.type == 'stop': await self._do_shutdown() @@ -156,7 +156,7 @@ def _trigger_flush(self): self._diagnostic_accumulator.record_events_in_batch(len(payload.events)) if len(payload.events) > 0 or not payload.summary.is_empty(): task = EventPayloadSendTask(self._http, self._config, self._formatter, payload, self._handle_response) - if self._flush_workers.execute(task.run): + if self._flush_workers.try_run(task.run): # The events have been handed off to a flush worker; clear them from our buffer. self._outbox.clear() else: @@ -169,15 +169,15 @@ def _send_and_reset_diagnostics(self): stats_event = self._diagnostic_accumulator.create_event_and_reset(dropped_event_count, self._deduplicated_contexts) self._deduplicated_contexts = 0 task = DiagnosticEventSendTask(self._http, self._config, stats_event) - self._diagnostic_flush_workers.execute(task.run) + self._diagnostic_flush_workers.try_run(task.run) async def _do_shutdown(self): self._flush_workers.stop() - await self._flush_workers.wait() + await self._flush_workers.drain() if self._diagnostic_flush_workers is not None: self._diagnostic_flush_workers.stop() - await self._diagnostic_flush_workers.wait() + await self._diagnostic_flush_workers.drain() await self._http.close() diff --git a/ldclient/testing/test_aio.py b/ldclient/testing/test_aio.py index 6bf8d8ab..9b352f6d 100644 --- a/ldclient/testing/test_aio.py +++ b/ldclient/testing/test_aio.py @@ -203,13 +203,13 @@ async def action(): # --------------------------------------------------------------------------- -# WorkerPool +# BoundedTaskSet # --------------------------------------------------------------------------- -class TestWorkerPoolParity: +class TestBoundedTaskSet: @pytest.mark.asyncio - async def test_async_saturation_returns_false(self): - pool = aio.AsyncWorkerPool(1, "test.pool") + async def test_saturation_returns_false(self): + tasks = aio.BoundedTaskSet(1) release = aio.AsyncEvent() started = aio.AsyncEvent() @@ -217,24 +217,29 @@ async def job(): started.set() await release.wait(2) - assert pool.execute(job) is True + async def noop(): + pass + + assert tasks.try_run(job) is True await started.wait(2) - assert pool.execute(lambda: None) is False + # Set is full (limit 1), so the next task is rejected rather than queued. + assert tasks.try_run(noop) is False release.set() - await pool.wait() + await tasks.drain() - async def noop(): - pass + # A slot is free again once the first task drained. + assert tasks.try_run(noop) is True + await tasks.drain() - assert pool.execute(noop) is True - await pool.wait() - pool.stop() + # After stop(), further work is rejected. + tasks.stop() + assert tasks.try_run(noop) is False @pytest.mark.asyncio - async def test_async_wait_returns_when_idle(self): - pool = aio.AsyncWorkerPool(2, "test.pool") - await pool.wait() - pool.stop() + async def test_drain_returns_when_idle(self): + tasks = aio.BoundedTaskSet(2) + await tasks.drain() + tasks.stop() # --------------------------------------------------------------------------- From 50b9e040818d6e0cf47fd2ce7ab2991fe6dd248b Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 30 Jul 2026 11:36:50 -0500 Subject: [PATCH 5/7] refactor: Drop section-heading comments in event_processor_common Remove the banner-style section headings; where they carried a useful note, capture it as a short class docstring instead. --- .../impl/events/event_processor_common.py | 20 ++++--------------- 1 file changed, 4 insertions(+), 16 deletions(-) diff --git a/ldclient/impl/events/event_processor_common.py b/ldclient/impl/events/event_processor_common.py index b8b60827..5f6c2595 100644 --- a/ldclient/impl/events/event_processor_common.py +++ b/ldclient/impl/events/event_processor_common.py @@ -30,10 +30,6 @@ ) from ldclient.migrations.tracker import MigrationOpEvent -# --------------------------------------------------------------------------- -# Shared event wrapper types -# --------------------------------------------------------------------------- - class DebugEvent: __slots__ = ['original_input'] @@ -53,11 +49,9 @@ def __init__(self, timestamp: int, context: Context): FlushPayload = namedtuple('FlushPayload', ['events', 'summary']) -# --------------------------------------------------------------------------- -# EventBuffer — in-memory accumulation buffer (no I/O) -# --------------------------------------------------------------------------- - class EventBuffer: + """In-memory buffer that accumulates events and their summary until flush. Performs no I/O.""" + def __init__(self, capacity: int): self._capacity = capacity self._events: List[Any] = [] @@ -91,11 +85,9 @@ def clear(self): self._summarizer.clear() -# --------------------------------------------------------------------------- -# EventOutputFormatter — pure data transform (no I/O) -# --------------------------------------------------------------------------- - class EventOutputFormatter: + """Transforms buffered events into the LaunchDarkly wire-format output payload. Performs no I/O.""" + def __init__(self, config: PrivateAttributesConfig): self._context_formatter = EventContextFormatter( config.all_attributes_private, config.private_attributes @@ -236,10 +228,6 @@ def _base_eval_props(self, e: EventInputEvaluation, kind: str) -> dict: return out -# --------------------------------------------------------------------------- -# EventDispatcherBase — pure event-handling logic shared by both dispatchers -# --------------------------------------------------------------------------- - class EventDispatcherBase: """ Pure event-handling methods shared by the sync and async EventDispatcher From 9ed47d165b328bfe9ea7d82a8f8911d9e8a483f4 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 30 Jul 2026 12:33:44 -0500 Subject: [PATCH 6/7] refactor: Rename BoundedTaskSet.wait, simplify job type, drop test section comments Rename drain() back to wait() (it awaits the in-flight tasks; it does not halt intake), simplify the job param to Callable[[], Coroutine], and remove banner-style section-heading comments from the async event/concurrency tests. --- ldclient/impl/aio/concurrency.py | 11 +++-- ldclient/impl/events/async_event_processor.py | 10 ++-- .../impl/events/test_async_event_processor.py | 39 +--------------- ldclient/testing/test_aio.py | 46 ++----------------- 4 files changed, 18 insertions(+), 88 deletions(-) diff --git a/ldclient/impl/aio/concurrency.py b/ldclient/impl/aio/concurrency.py index beafd56b..fd692df9 100644 --- a/ldclient/impl/aio/concurrency.py +++ b/ldclient/impl/aio/concurrency.py @@ -253,15 +253,15 @@ async def _run(self): class BoundedTaskSet: """Runs up to ``limit`` coroutines concurrently as background tasks. When the limit is reached, ``try_run`` rejects new work (returning False) rather than - queuing it, so callers can apply their own backpressure. ``drain`` awaits all - in-flight tasks; ``stop`` prevents any further work from being accepted.""" + queuing it, so callers can apply their own backpressure. ``wait`` awaits the + tasks currently in flight; ``stop`` prevents any further work from being accepted.""" def __init__(self, limit: int): self._limit = limit self._tasks: Set[asyncio.Task] = set() self._accepting = True - def try_run(self, job: Callable[[], Coroutine[Any, Any, Any]]) -> bool: + def try_run(self, job: Callable[[], Coroutine]) -> bool: """Runs ``job()`` as a background task if fewer than ``limit`` tasks are already running and the set is still accepting work. Returns True if the task was started, or False if the set is full or has been stopped.""" @@ -277,8 +277,9 @@ def _on_done(self, task: asyncio.Task) -> None: if not task.cancelled() and task.exception() is not None: log.warning('Unhandled exception in background task', exc_info=task.exception()) - async def drain(self) -> None: - """Waits for all currently running tasks to complete.""" + async def wait(self) -> None: + """Waits for the tasks currently running to complete. This does not stop + new tasks from being added; call ``stop`` first to prevent further work.""" await asyncio.gather(*self._tasks, return_exceptions=True) def stop(self) -> None: diff --git a/ldclient/impl/events/async_event_processor.py b/ldclient/impl/events/async_event_processor.py index ac9b2a5a..ae5881a1 100644 --- a/ldclient/impl/events/async_event_processor.py +++ b/ldclient/impl/events/async_event_processor.py @@ -134,12 +134,12 @@ async def _run_main_loop(self): self._send_and_reset_diagnostics() elif message.type == 'flush_and_wait': self._trigger_flush() - await self._flush_workers.drain() + await self._flush_workers.wait() message.param.set() elif message.type == 'test_sync': - await self._flush_workers.drain() + await self._flush_workers.wait() if self._diagnostic_flush_workers is not None: - await self._diagnostic_flush_workers.drain() + await self._diagnostic_flush_workers.wait() message.param.set() elif message.type == 'stop': await self._do_shutdown() @@ -173,11 +173,11 @@ def _send_and_reset_diagnostics(self): async def _do_shutdown(self): self._flush_workers.stop() - await self._flush_workers.drain() + await self._flush_workers.wait() if self._diagnostic_flush_workers is not None: self._diagnostic_flush_workers.stop() - await self._diagnostic_flush_workers.drain() + await self._diagnostic_flush_workers.wait() await self._http.close() diff --git a/ldclient/testing/impl/events/test_async_event_processor.py b/ldclient/testing/impl/events/test_async_event_processor.py index 780513c7..3c6abd9c 100644 --- a/ldclient/testing/impl/events/test_async_event_processor.py +++ b/ldclient/testing/impl/events/test_async_event_processor.py @@ -41,11 +41,6 @@ timestamp = 10000 -# --------------------------------------------------------------------------- -# Mock aiohttp session (mirrors stub_util.MockHttp for the aiohttp surface -# used by AsyncHTTPTransport) -# --------------------------------------------------------------------------- - class MockAioResponse: def __init__(self, status: int, headers: dict): self.status = status @@ -67,6 +62,8 @@ async def __aexit__(self, exc_type, exc_value, traceback) -> bool: class MockAioHttp: + """Mock aiohttp session mirroring stub_util.MockHttp for the aiohttp surface used by AsyncHTTPTransport.""" + def __init__(self): self._recorded_requests: List[Tuple[Any, Any]] = [] self._response_status = 200 @@ -103,10 +100,6 @@ def reset(self): self._recorded_requests = [] -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- - @asynccontextmanager async def make_processor(mock_http: MockAioHttp, **kwargs): kwargs.setdefault('diagnostic_opt_out', True) @@ -128,10 +121,6 @@ async def flush_and_get_events(ep: DefaultAsyncEventProcessor, mock_http: MockAi return json.loads(mock_http.request_data) -# --------------------------------------------------------------------------- -# Event payload tests -# --------------------------------------------------------------------------- - async def test_identify_event_is_queued(): mock_http = MockAioHttp() async with make_processor(mock_http) as ep: @@ -174,10 +163,6 @@ async def test_custom_event_is_queued(): assert output[1]['metricValue'] == 1.5 -# --------------------------------------------------------------------------- -# flush_and_wait -# --------------------------------------------------------------------------- - async def test_flush_and_wait_delivers_events_and_returns_true(): mock_http = MockAioHttp() async with make_processor(mock_http) as ep: @@ -250,10 +235,6 @@ async def test_stop_flushes_remaining_events(): assert output[0]['kind'] == 'identify' -# --------------------------------------------------------------------------- -# Header tests -# --------------------------------------------------------------------------- - async def test_sdk_key_is_sent(): mock_http = MockAioHttp() async with make_processor(mock_http, sdk_key='SDK_KEY') as ep: @@ -299,10 +280,6 @@ async def test_event_payload_id_changes_between_requests(): assert first_payload_id != second_payload_id -# --------------------------------------------------------------------------- -# Diagnostics -# --------------------------------------------------------------------------- - async def test_init_diagnostic_event_sent(): mock_http = MockAioHttp() async with make_processor(mock_http, diagnostic_opt_out=False) as ep: @@ -355,10 +332,6 @@ async def test_event_schema_not_set_on_diagnostic_send(): assert mock_http.request_headers.get('X-LaunchDarkly-Event-Schema') is None -# --------------------------------------------------------------------------- -# HTTP error handling -# --------------------------------------------------------------------------- - async def verify_unrecoverable_http_error(status: int): mock_http = MockAioHttp() async with make_processor(mock_http, sdk_key='SDK_KEY') as ep: @@ -409,10 +382,6 @@ async def test_will_still_send_after_500_error(): await verify_recoverable_http_error(500) -# --------------------------------------------------------------------------- -# Inbox capacity -# --------------------------------------------------------------------------- - async def test_does_not_block_on_full_inbox(): config = AsyncConfig("fake_sdk_key", events_max_pending=1) # this sets the size of both the inbox and the outbox to 1 ep_inbox_holder = [None] @@ -444,10 +413,6 @@ async def event_consumer(): assert had_no_more -# --------------------------------------------------------------------------- -# Compression -# --------------------------------------------------------------------------- - async def test_event_payload_is_gzip_compressed_when_enabled(): mock_http = MockAioHttp() async with make_processor(mock_http, enable_event_compression=True) as ep: diff --git a/ldclient/testing/test_aio.py b/ldclient/testing/test_aio.py index 9b352f6d..544319bf 100644 --- a/ldclient/testing/test_aio.py +++ b/ldclient/testing/test_aio.py @@ -28,10 +28,6 @@ async def _async_wait_until(predicate, timeout=2.0): await asyncio.sleep(0.01) -# --------------------------------------------------------------------------- -# Event -# --------------------------------------------------------------------------- - class TestEventParity: @pytest.mark.asyncio async def test_async_set_clear_is_set(self): @@ -63,10 +59,6 @@ async def setter(): await task -# --------------------------------------------------------------------------- -# Lock -# --------------------------------------------------------------------------- - class TestLockParity: @pytest.mark.asyncio async def test_async_lock_context_manager(self): @@ -94,10 +86,6 @@ async def work(): assert counter['max_concurrent'] == 1 -# --------------------------------------------------------------------------- -# Queue -# --------------------------------------------------------------------------- - class TestQueueParity: @pytest.mark.asyncio async def test_async_put_get_and_empty(self): @@ -129,10 +117,6 @@ async def test_async_put_nowait_raises_queue_full_at_capacity(self): assert await q.get(block=False) == 'a' -# --------------------------------------------------------------------------- -# RepeatingTask -# --------------------------------------------------------------------------- - class TestRepeatingTaskParity: @pytest.mark.asyncio async def test_async_fires_repeatedly_then_stops(self): @@ -202,10 +186,6 @@ async def action(): task.stop() -# --------------------------------------------------------------------------- -# BoundedTaskSet -# --------------------------------------------------------------------------- - class TestBoundedTaskSet: @pytest.mark.asyncio async def test_saturation_returns_false(self): @@ -225,27 +205,23 @@ async def noop(): # Set is full (limit 1), so the next task is rejected rather than queued. assert tasks.try_run(noop) is False release.set() - await tasks.drain() + await tasks.wait() - # A slot is free again once the first task drained. + # A slot is free again once the first task finished. assert tasks.try_run(noop) is True - await tasks.drain() + await tasks.wait() # After stop(), further work is rejected. tasks.stop() assert tasks.try_run(noop) is False @pytest.mark.asyncio - async def test_drain_returns_when_idle(self): + async def test_wait_returns_when_idle(self): tasks = aio.BoundedTaskSet(2) - await tasks.drain() + await tasks.wait() tasks.stop() -# --------------------------------------------------------------------------- -# TaskRunner -# --------------------------------------------------------------------------- - class TestTaskRunnerParity: @pytest.mark.asyncio async def test_async_spawn_runs_function(self): @@ -302,10 +278,6 @@ async def stubborn(): pass -# --------------------------------------------------------------------------- -# spawn_handle / join_handle -# --------------------------------------------------------------------------- - class TestHandleParity: @pytest.mark.asyncio async def test_async_spawn_and_join(self): @@ -366,10 +338,6 @@ async def slow(): await joiner -# --------------------------------------------------------------------------- -# Callback scheduler -# --------------------------------------------------------------------------- - class TestCallbackSchedulerParity: @pytest.mark.asyncio async def test_async_call_schedules_coroutine_with_args(self): @@ -409,10 +377,6 @@ async def boom(): await asyncio.sleep(0.05) # let the done callback run -# --------------------------------------------------------------------------- -# HTTP transport -# --------------------------------------------------------------------------- - class _FakeResponse: status = 200 headers: dict = {} From 8c14ed7e62109c0341d518c81ed9c84671f93920 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 30 Jul 2026 12:50:22 -0500 Subject: [PATCH 7/7] refactor: Share CURRENT_EVENT_SCHEMA between sync and async event processors Move the event schema version constant into event_processor_common so both processors advertise the same X-LaunchDarkly-Event-Schema and can't drift. --- ldclient/impl/events/async_event_processor.py | 4 ++-- ldclient/impl/events/event_processor.py | 4 ++-- ldclient/impl/events/event_processor_common.py | 3 +++ 3 files changed, 7 insertions(+), 4 deletions(-) diff --git a/ldclient/impl/events/async_event_processor.py b/ldclient/impl/events/async_event_processor.py index ae5881a1..996bc72c 100644 --- a/ldclient/impl/events/async_event_processor.py +++ b/ldclient/impl/events/async_event_processor.py @@ -24,6 +24,7 @@ from ldclient.impl.aio.transport import AsyncHTTPTransport from ldclient.impl.events.diagnostics import create_diagnostic_init from ldclient.impl.events.event_processor_common import ( + CURRENT_EVENT_SCHEMA, EventBuffer, EventDispatcherBase, EventOutputFormatter @@ -39,7 +40,6 @@ from ldclient.interfaces import AsyncEventProcessor __MAX_FLUSH_CONCURRENCY__ = 5 -__CURRENT_EVENT_SCHEMA__ = 4 EventProcessorMessage = namedtuple('EventProcessorMessage', ['type', 'param']) @@ -268,7 +268,7 @@ async def _post_events_with_retry(http_client: AsyncHTTPTransport, config: Async hdrs['Content-Encoding'] = 'gzip' if payload_id: - hdrs['X-LaunchDarkly-Event-Schema'] = str(__CURRENT_EVENT_SCHEMA__) + hdrs['X-LaunchDarkly-Event-Schema'] = str(CURRENT_EVENT_SCHEMA) hdrs['X-LaunchDarkly-Payload-ID'] = payload_id can_retry = True context = "posting %s" % events_description diff --git a/ldclient/impl/events/event_processor.py b/ldclient/impl/events/event_processor.py index f9ee3ddb..20cf03a4 100644 --- a/ldclient/impl/events/event_processor.py +++ b/ldclient/impl/events/event_processor.py @@ -16,6 +16,7 @@ from ldclient.config import Config from ldclient.impl.events.diagnostics import create_diagnostic_init from ldclient.impl.events.event_processor_common import ( + CURRENT_EVENT_SCHEMA, EventBuffer, EventDispatcherBase, EventOutputFormatter @@ -34,7 +35,6 @@ from ldclient.interfaces import EventProcessor __MAX_FLUSH_THREADS__ = 5 -__CURRENT_EVENT_SCHEMA__ = 4 EventProcessorMessage = namedtuple('EventProcessorMessage', ['type', 'param']) @@ -250,7 +250,7 @@ def _post_events_with_retry(http_client, config, uri, payload_id, body, events_d hdrs['Content-Encoding'] = 'gzip' if payload_id: - hdrs['X-LaunchDarkly-Event-Schema'] = str(__CURRENT_EVENT_SCHEMA__) + hdrs['X-LaunchDarkly-Event-Schema'] = str(CURRENT_EVENT_SCHEMA) hdrs['X-LaunchDarkly-Payload-ID'] = payload_id can_retry = True context = "posting %s" % events_description diff --git a/ldclient/impl/events/event_processor_common.py b/ldclient/impl/events/event_processor_common.py index 5f6c2595..eb00bae5 100644 --- a/ldclient/impl/events/event_processor_common.py +++ b/ldclient/impl/events/event_processor_common.py @@ -30,6 +30,9 @@ ) from ldclient.migrations.tracker import MigrationOpEvent +# Analytics event schema version sent to LaunchDarkly via the X-LaunchDarkly-Event-Schema header. +CURRENT_EVENT_SCHEMA = 4 + class DebugEvent: __slots__ = ['original_input']