From 8d8308d61476c4a086959d93910a09c930909800 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Mon, 27 Jul 2026 17:22:18 -0500 Subject: [PATCH 01/11] 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 f1f51e9bcba80498c0f93a65c32948329e36d2e3 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Tue, 28 Jul 2026 09:08:03 -0500 Subject: [PATCH 02/11] 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 d300c4e5fa1e11e1d2cfb60dadeb20d1ea74e5be Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Wed, 29 Jul 2026 13:38:26 -0500 Subject: [PATCH 03/11] 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 6b58a116e833beccfe74e62d223d83f454b679a5 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 30 Jul 2026 11:18:08 -0500 Subject: [PATCH 04/11] 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 6f75754a..4633dc99 100644 --- a/ldclient/testing/test_aio.py +++ b/ldclient/testing/test_aio.py @@ -205,13 +205,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() @@ -219,24 +219,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 f69f0402b154859dcd5f14d215bcb4c31a0733c7 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 30 Jul 2026 11:36:50 -0500 Subject: [PATCH 05/11] 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 e26d1ca6545515c49b375ae91b674fd2a10651ac Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 30 Jul 2026 12:33:44 -0500 Subject: [PATCH 06/11] 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 4633dc99..2daf4d72 100644 --- a/ldclient/testing/test_aio.py +++ b/ldclient/testing/test_aio.py @@ -30,10 +30,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): @@ -65,10 +61,6 @@ async def setter(): await task -# --------------------------------------------------------------------------- -# Lock -# --------------------------------------------------------------------------- - class TestLockParity: @pytest.mark.asyncio async def test_async_lock_context_manager(self): @@ -96,10 +88,6 @@ async def work(): assert counter['max_concurrent'] == 1 -# --------------------------------------------------------------------------- -# Queue -# --------------------------------------------------------------------------- - class TestQueueParity: @pytest.mark.asyncio async def test_async_put_get_and_empty(self): @@ -131,10 +119,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): @@ -204,10 +188,6 @@ async def action(): task.stop() -# --------------------------------------------------------------------------- -# BoundedTaskSet -# --------------------------------------------------------------------------- - class TestBoundedTaskSet: @pytest.mark.asyncio async def test_saturation_returns_false(self): @@ -227,27 +207,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): @@ -304,10 +280,6 @@ async def stubborn(): pass -# --------------------------------------------------------------------------- -# spawn_handle / join_handle -# --------------------------------------------------------------------------- - class TestHandleParity: @pytest.mark.asyncio async def test_async_spawn_and_join(self): @@ -368,10 +340,6 @@ async def slow(): await joiner -# --------------------------------------------------------------------------- -# Callback scheduler -# --------------------------------------------------------------------------- - class TestCallbackSchedulerParity: @pytest.mark.asyncio async def test_async_call_schedules_coroutine_with_args(self): @@ -411,10 +379,6 @@ async def boom(): await asyncio.sleep(0.05) # let the done callback run -# --------------------------------------------------------------------------- -# HTTP transport -# --------------------------------------------------------------------------- - class _FakeResponse: status = 200 headers: dict = {} From 57419caeef7ea3e092a45e42732cf60b1f77c42a Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 30 Jul 2026 12:50:22 -0500 Subject: [PATCH 07/11] 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'] From 8677f88cc7df00106c65e1aa792027d663a1469d Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Fri, 31 Jul 2026 11:30:46 -0500 Subject: [PATCH 08/11] fix: Ensure flush_and_wait delivers the batch before reporting success _trigger_flush now returns whether it handed the batch to a worker; the flush_and_wait handler retries (waiting for a free worker) until the batch is handed off, so a saturated worker pool no longer causes a false success. --- ldclient/impl/events/async_event_processor.py | 21 ++++++++---- .../impl/events/test_async_event_processor.py | 32 ++++++++++++++++++- 2 files changed, 46 insertions(+), 7 deletions(-) diff --git a/ldclient/impl/events/async_event_processor.py b/ldclient/impl/events/async_event_processor.py index 996bc72c..49cdfc5e 100644 --- a/ldclient/impl/events/async_event_processor.py +++ b/ldclient/impl/events/async_event_processor.py @@ -133,7 +133,12 @@ async def _run_main_loop(self): elif message.type == 'diagnostic': self._send_and_reset_diagnostics() elif message.type == 'flush_and_wait': - self._trigger_flush() + # Ensure the buffered batch is actually handed to a worker + # before reporting success; under saturation the first + # trigger may leave the events buffered, so wait for a free + # worker and retry. + while not self._trigger_flush(): + await self._flush_workers.wait() await self._flush_workers.wait() message.param.set() elif message.type == 'test_sync': @@ -148,9 +153,12 @@ async def _run_main_loop(self): except Exception: log.error('Unhandled exception in event processor', exc_info=True) - def _trigger_flush(self): + def _trigger_flush(self) -> bool: + """Hands the buffered events to a flush worker. Returns True if handed + off (or there was nothing to flush), or False if all workers are busy + and the events were left buffered for a later attempt.""" if self._disabled: - return + return True payload = self._outbox.get_payload() if self._diagnostic_accumulator: self._diagnostic_accumulator.record_events_in_batch(len(payload.events)) @@ -159,9 +167,10 @@ def _trigger_flush(self): 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: - # We're already at our limit of concurrent flushes; leave the events in the buffer. - pass + return True + # We're already at our limit of concurrent flushes; leave the events in the buffer. + return False + return True def _send_and_reset_diagnostics(self): if self._diagnostic_accumulator is not None and self._diagnostic_flush_workers is not None: diff --git a/ldclient/testing/impl/events/test_async_event_processor.py b/ldclient/testing/impl/events/test_async_event_processor.py index 3c6abd9c..94b3035e 100644 --- a/ldclient/testing/impl/events/test_async_event_processor.py +++ b/ldclient/testing/impl/events/test_async_event_processor.py @@ -14,13 +14,16 @@ from contextlib import asynccontextmanager from email.utils import formatdate from typing import Any, List, Optional, Tuple +from unittest.mock import MagicMock import pytest from ldclient.async_config import AsyncConfig from ldclient.context import Context +from ldclient.impl.aio.concurrency import AsyncQueue from ldclient.impl.events.async_event_processor import ( - DefaultAsyncEventProcessor + DefaultAsyncEventProcessor, + EventDispatcher ) from ldclient.impl.events.diagnostics import ( _DiagnosticAccumulator, @@ -188,6 +191,33 @@ async def test_flush_and_wait_returns_false_on_timeout(): assert delivered is False +async def test_trigger_flush_reports_whether_batch_handed_off(): + # flush_and_wait relies on _trigger_flush reporting whether the batch was + # actually handed to a worker; when the pool is saturated it must report + # False (events left buffered) so the caller retries rather than falsely + # reporting success. + mock_http = MockAioHttp() + config = AsyncConfig(sdk_key='SDK_KEY', diagnostic_opt_out=True) + dispatcher = EventDispatcher(AsyncQueue(config.events_max_pending), config, mock_http) + try: + # Nothing buffered -> handed off (nothing to do). + assert dispatcher._trigger_flush() is True + + dispatcher._outbox.add_event(EventInputIdentify(timestamp, context)) + + # Saturated pool: try_run rejects, events stay buffered. + dispatcher._flush_workers = MagicMock() + dispatcher._flush_workers.try_run = MagicMock(return_value=False) + assert dispatcher._trigger_flush() is False + + # A free slot: hands off and clears the buffer. + dispatcher._flush_workers.try_run = MagicMock(return_value=True) + assert dispatcher._trigger_flush() is True + assert dispatcher._outbox.get_payload().events == [] + finally: + await dispatcher._runner.stop_all() + + 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 27f38979a3a4b86a159083efa2cf808c5961979f Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Tue, 4 Aug 2026 10:56:36 -0600 Subject: [PATCH 09/11] docs: Simplify event processor comment wording --- ldclient/impl/events/async_event_processor.py | 7 +++---- ldclient/impl/events/event_processor_common.py | 2 +- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/ldclient/impl/events/async_event_processor.py b/ldclient/impl/events/async_event_processor.py index 49cdfc5e..085d1fb1 100644 --- a/ldclient/impl/events/async_event_processor.py +++ b/ldclient/impl/events/async_event_processor.py @@ -133,10 +133,9 @@ async def _run_main_loop(self): elif message.type == 'diagnostic': self._send_and_reset_diagnostics() elif message.type == 'flush_and_wait': - # Ensure the buffered batch is actually handed to a worker - # before reporting success; under saturation the first - # trigger may leave the events buffered, so wait for a free - # worker and retry. + # Do not report success until a worker takes the batch. + # When all workers are busy, the first trigger leaves the + # events buffered, so wait for a free worker and try again. while not self._trigger_flush(): await self._flush_workers.wait() await self._flush_workers.wait() diff --git a/ldclient/impl/events/event_processor_common.py b/ldclient/impl/events/event_processor_common.py index eb00bae5..95ea278f 100644 --- a/ldclient/impl/events/event_processor_common.py +++ b/ldclient/impl/events/event_processor_common.py @@ -89,7 +89,7 @@ def clear(self): class EventOutputFormatter: - """Transforms buffered events into the LaunchDarkly wire-format output payload. Performs no I/O.""" + """Transforms buffered events into the output payload sent to LaunchDarkly. Performs no I/O.""" def __init__(self, config: PrivateAttributesConfig): self._context_formatter = EventContextFormatter( From 623faa001bdc449295771b68ee48cc1f2d5ff5ea Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Tue, 4 Aug 2026 11:26:58 -0600 Subject: [PATCH 10/11] docs: Simplify async concurrency module and scheduler docstrings --- ldclient/impl/aio/concurrency.py | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/ldclient/impl/aio/concurrency.py b/ldclient/impl/aio/concurrency.py index fd692df9..b4ac4460 100644 --- a/ldclient/impl/aio/concurrency.py +++ b/ldclient/impl/aio/concurrency.py @@ -1,10 +1,10 @@ """ -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 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. +Async concurrency helpers used by the async data source, event processor, and +data system. Each helper wraps one piece of asyncio setup so callers do not +repeat it: timeout-aware waits, queue timeout/capacity exceptions, a repeating +task, and a bounded task set. The sync code uses the standard-library and SDK +equivalents (``threading.Event``/``Lock``, ``queue.Queue``, ``RepeatingTask``, +``FixedThreadPool``) directly, so it has no matching helpers. """ import asyncio @@ -130,9 +130,9 @@ async def join_handle(handle: TaskHandle, timeout: float) -> None: class AsyncCallbackScheduler: - """Bridges sync notification paths to async callbacks: ``call`` schedules - a coroutine callback onto the event loop captured at construction time, - logging any unhandled exception. Safe to invoke from any thread.""" + """Schedules a coroutine callback onto the event loop that was running when + this object was created. ``call`` runs the callback and logs any unhandled + exception. Safe to call from any thread.""" def __init__(self): self._loop = asyncio.get_running_loop() From 1dfcdfd7dd6e0f4bdf15178c76c53f293bb135f9 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Tue, 4 Aug 2026 17:08:24 -0600 Subject: [PATCH 11/11] fix: Complete async event processor shutdown and flush buffered events Two shutdown bugs on the async event processor's stop path: - If _do_shutdown raised, the dispatcher logged it and continued the loop without setting the stop reply or returning, so stop() (which waits on that reply with no timeout) hung forever. Guard the stop branch so the reply is always set and the loop exits. - _do_shutdown never drained the outbox, so buffered events were lost on shutdown if the pre-stop flush was dropped (inbox full) or left buffered (workers saturated). Hand off the outbox with retry before stopping the workers. Adds regression tests for both. The sync event processor has the same two issues; tracked as a follow-up. --- ldclient/impl/events/async_event_processor.py | 10 ++++- .../impl/events/test_async_event_processor.py | 41 ++++++++++++++++++- 2 files changed, 48 insertions(+), 3 deletions(-) diff --git a/ldclient/impl/events/async_event_processor.py b/ldclient/impl/events/async_event_processor.py index 085d1fb1..e09b1c21 100644 --- a/ldclient/impl/events/async_event_processor.py +++ b/ldclient/impl/events/async_event_processor.py @@ -146,7 +146,10 @@ async def _run_main_loop(self): await self._diagnostic_flush_workers.wait() message.param.set() elif message.type == 'stop': - await self._do_shutdown() + try: + await self._do_shutdown() + except Exception: + log.error('Error during event processor shutdown', exc_info=True) message.param.set() return except Exception: @@ -180,6 +183,11 @@ def _send_and_reset_diagnostics(self): self._diagnostic_flush_workers.try_run(task.run) async def _do_shutdown(self): + # Deliver any still-buffered events before shutting down. Retry the + # hand-off while all flush workers are busy (like flush_and_wait), then + # stop the pool and wait for the in-flight flushes to finish. + while not self._trigger_flush(): + await self._flush_workers.wait() self._flush_workers.stop() await self._flush_workers.wait() diff --git a/ldclient/testing/impl/events/test_async_event_processor.py b/ldclient/testing/impl/events/test_async_event_processor.py index 94b3035e..8ffc41b0 100644 --- a/ldclient/testing/impl/events/test_async_event_processor.py +++ b/ldclient/testing/impl/events/test_async_event_processor.py @@ -20,10 +20,11 @@ from ldclient.async_config import AsyncConfig from ldclient.context import Context -from ldclient.impl.aio.concurrency import AsyncQueue +from ldclient.impl.aio.concurrency import AsyncEvent, AsyncQueue from ldclient.impl.events.async_event_processor import ( DefaultAsyncEventProcessor, - EventDispatcher + EventDispatcher, + EventProcessorMessage ) from ldclient.impl.events.diagnostics import ( _DiagnosticAccumulator, @@ -218,6 +219,42 @@ async def test_trigger_flush_reports_whether_batch_handed_off(): await dispatcher._runner.stop_all() +@pytest.mark.asyncio +async def test_stop_completes_even_if_shutdown_raises(): + # If _do_shutdown raises, the dispatcher must still set the stop reply and + # exit the loop; otherwise stop() would wait on that reply forever. + mock_http = MockAioHttp() + config = AsyncConfig(sdk_key='SDK_KEY', diagnostic_opt_out=True) + inbox = AsyncQueue(config.events_max_pending) + dispatcher = EventDispatcher(inbox, config, mock_http) + + async def boom(): + raise RuntimeError("shutdown failed") + dispatcher._do_shutdown = boom + + reply = AsyncEvent() + await inbox.put(EventProcessorMessage('stop', reply)) + # Would hang without the fix; wait_for turns a hang into a test failure. + await asyncio.wait_for(reply.wait(), 2) + + +@pytest.mark.asyncio +async def test_shutdown_flushes_buffered_events(): + # _do_shutdown must drain the outbox so buffered events are delivered on a + # clean shutdown, even when no flush reached the dispatcher beforehand. + mock_http = MockAioHttp() + config = AsyncConfig(sdk_key='SDK_KEY', diagnostic_opt_out=True) + inbox = AsyncQueue(config.events_max_pending) + dispatcher = EventDispatcher(inbox, config, mock_http) + dispatcher._outbox.add_event(EventInputIdentify(timestamp, context)) + + reply = AsyncEvent() + await inbox.put(EventProcessorMessage('stop', reply)) + await asyncio.wait_for(reply.wait(), 2) + + assert mock_http.request_data is not None + + async def test_two_events_for_same_context_only_produce_one_index_event(): mock_http = MockAioHttp() async with make_processor(mock_http) as ep: