feat: Add async event processor

[jsonbailey]

Adds the async analytics event delivery component for the async SDK client, extracted from the async SDK implementation branch (SDK-60).

Stacked on #471 (AsyncConfig) — review that first. Until #471 merges this PR shows its commits too; after merge a rebase drops them out.

What's here

• AsyncEventProcessor interface + DefaultAsyncEventProcessor concrete implementation, mirroring the sync EventProcessor/DefaultEventProcessor split. send_event/flush are sync; stop is a coroutine.
• Shared sans-I/O helpers extracted into event_processor_common.py so the sync and async processors share buffering, output formatting, and dispatch logic.
• Tests mirroring the sync DefaultEventProcessor suite over an injected mock aiohttp session.

SDK-2769

feat: Add async event processor

---

Note

Medium Risk
Changes analytics delivery and refactors shared sync dispatch paths; behavior is intended to match sync with broad tests, but incorrect flush/backpressure or shared-base regressions could drop or mis-deliver events.

Overview
Introduces async analytics event delivery for AsyncConfig / AsyncLDClient: DefaultAsyncEventProcessor with inbox-driven EventDispatcher, periodic flush/context/diagnostic timers, gzip POSTs via AsyncHTTPTransport, and async stop / flush_and_wait semantics aligned with the sync processor.

Shared logic moves into event_processor_common as EventDispatcherBase (_process_event, indexing/dedup, debug events, _handle_response) plus a single CURRENT_EVENT_SCHEMA; the sync EventDispatcher now subclasses it instead of duplicating that code.

Concurrency:AsyncWorkerPool becomes BoundedTaskSet (try_run / gather-based wait) to cap concurrent flush POSTs; _trigger_flush returns whether a batch was handed off so flush_and_wait and shutdown can retry when all workers are busy. Tests cover shutdown flush, stop-on-shutdown-error, and a flush_and_wait livelock regression when workers finish in the same event-loop batch.

Adds a large async test suite mirroring sync DefaultEventProcessor behavior (headers, diagnostics, HTTP error handling, full inbox, compression).

^{Reviewed by Cursor Bugbot for commit 2ea75f5. Bugbot is set up for automated code reviews on this repo. Configure here.}

[jsonbailey]

The AsyncWorkerPool was previously created to mirror the FixedThreadPool but some of the naming and params didn't seem to align with the async code.

[jsonbailey]

Shutdown event-delivery parity (sync vs. async) — decision needed

While addressing the flush_and_wait saturation finding, Bugbot flagged a related gap: stop() promises to deliver all pending events, but _do_shutdown never re-flushes the outbox. The pre-stop flush() can be dropped when the inbox is full, or left buffered when the flush workers are saturated — so those events can be abandoned on shutdown.

The proposed async fix is to have _do_shutdown guarantee-flush the outbox before draining workers (reusing the same retry-until-handed-off logic that now backs flush_and_wait).

The catch: this gap exists identically in the sync EventProcessor — its stop() / _do_shutdown have the same structure and can drop events the same way. Fixing only the async side would make async's shutdown more robust than sync's (a behavioral divergence), whereas today they match.

Question: should we fix the sync side too, and if so —

• (a) bundle the sync + async fix together (here or a combined change), keeping the two in parity, or
• (b) land the async fix here and do sync in a separate PR/follow-up?

(Leaning toward keeping this PR async-scoped and handling sync separately, but flagging for a call before applying.)

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

There are 2 total unresolved issues (including 1 from previous review).

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 623faa0. Configure here.}

[kinyoklion]

This is a comment from Claude, an AI code reviewer. A human requested and reviewed this comment before it was posted.

Problem

The flush_and_wait handler can stop the event loop permanently. The dispatcher then uses approximately 100% CPU. All tasks on the application's event loop stop. Timers on that loop do not fire.

Cause

Two behaviors cause the problem:

1. BoundedTaskSet releases a slot only in _on_done. The event loop runs _on_done later, through call_soon. A task can be complete while the set continues to count it against the limit.
2. BoundedTaskSet.wait() awaits asyncio.gather(...). When all tasks in the set are complete, or when the set is empty, gather returns a future that is already complete. An await on a complete future does not give control back to the event loop.

The dispatcher uses this loop in the flush_and_wait branch of _run_main_loop:

whilenotself._trigger_flush():
awaitself._flush_workers.wait()

This sequence causes the stop:

1. The 5 POST tasks become complete in one event-loop batch.
2. The dispatcher receives the flush_and_wait message in the same batch.
3. try_run returns False, because the set counts the 5 complete tasks.
4. wait() returns immediately. It does not give control to the event loop.
5. The _on_done callbacks cannot run. The loop repeats without end.

With a real transport, this interleaving is a race. Socket-read completions and the inbox put only have to land in one batch. The sync SDK does not have this problem, because FixedThreadPool decreases its busy count synchronously in the worker thread.

Tests

The two tests below assert the correct behavior:

• Test 1 shows the defect at the BoundedTaskSet level. It is small and fast. It can go into TestBoundedTaskSet in ldclient/testing/test_aio.py.
• Test 2 shows the defect end-to-end on DefaultAsyncEventProcessor. It can go into ldclient/testing/impl/events/test_async_event_processor.py.

Results on this branch (623faa0): the 2 tests fail in approximately 3 seconds. They do not hang the test run. A stopped event loop cannot fire its own asyncio timeouts. Because of this, test 2 uses a watchdog thread. The watchdog finds the stop after 3 seconds and clears the task set. The test then fails with a clear assertion, and CI does not hang.

importasyncioimportthreadingimportpytestfromldclient.async_configimportAsyncConfigfromldclient.contextimportContextfromldclient.impl.aio.concurrencyimportAsyncEvent, BoundedTaskSetfromldclient.impl.events.async_event_processorimport (
DefaultAsyncEventProcessor,
EventDispatcher,
EventProcessorMessage
)
fromldclient.impl.events.typesimportEventInputIdentifyfromldclient.testing.impl.events.test_async_event_processorimport (
MockAioHttp,
MockAioResponse
)
pytestmark=pytest.mark.asynciocontext=Context.builder('userkey').name('Red').build()
timestamp=10000asyncdeftest_retry_pattern_makes_progress_when_a_task_finished_in_the_same_batch():
tasks=BoundedTaskSet(1)
finished=asyncio.Event()
asyncdefjob():
finished.set()
asserttasks.try_run(job) isTrue# A raw asyncio.Event wakes this coroutine in the same loop batch in which# the job's task finished, *before* the task's done-callback has run.# This is exactly the window in which the dispatcher's flush_and_wait# handler can observe the set (a task completion and an inbox put landing# in one batch).awaitfinished.wait()
# Desired behavior: the retry pattern used by EventDispatcher._run_main_loop# while not try_run(...): await wait()# must make progress here. Bounded to 100 iterations so the defect shows# up as a clean assertion failure rather than a wedged test run.asyncdefnoop():
passaccepted=Falsefor_inrange(100):
iftasks.try_run(noop):
accepted=Truebreakawaittasks.wait()
assertaccepted, (
"try_run never freed capacity: wait() returned without yielding to the ""event loop, so the done-callback that reaps the finished task can ""never run; the equivalent unbounded loop in the event dispatcher ""spins forever"
)
awaittasks.wait()
asyncdeftest_flush_and_wait_completes_when_in_flight_posts_finish_together():
classGatedMockAioHttp(MockAioHttp):
"""Requests park until the gate is set, then complete without yielding."""def__init__(self):
super().__init__()
self.gate=asyncio.Event()
defrequest(self, method, uri, headers=None, data=None, timeout=None, proxy=None):
self._recorded_requests.append((headers, data))
outer=selfclass_Ctx:
asyncdef__aenter__(self):
awaitouter.gate.wait()
returnMockAioResponse(200, {})
asyncdef__aexit__(self, exc_type, exc_value, traceback):
returnFalsereturn_Ctx()
# Keep a handle on the dispatcher (DefaultAsyncEventProcessor discards it)# so the watchdog below can recover the loop if it wedges.dispatcher_holder= []
defcapture_dispatcher(inbox, config, http, diagnostic_accumulator):
dispatcher=EventDispatcher(inbox, config, http, diagnostic_accumulator)
dispatcher_holder.append(dispatcher)
returndispatchermock_http=GatedMockAioHttp()
config=AsyncConfig(sdk_key='SDK_KEY', diagnostic_opt_out=True)
ep=DefaultAsyncEventProcessor(config, mock_http, dispatcher_class=capture_dispatcher)
try:
# Saturate all 5 flush workers with parked POSTs.foriinrange(5):
ep.send_event(EventInputIdentify(timestamp, Context.create('user%d'%i)))
ep.flush()
deadline=asyncio.get_running_loop().time() +2whilelen(mock_http.recorded_requests) <5:
assertasyncio.get_running_loop().time() <deadline, 'workers never saturated'awaitasyncio.sleep(0.01)
# Buffer one more event so the awaited flush has work to hand off.ep.send_event(EventInputIdentify(timestamp, context))
awaitasyncio.sleep(0.05)
# The race, made deterministic: complete all in-flight POSTs and# enqueue the flush_and_wait message inside one loop callback, so the# dispatcher dequeues it in the same batch in which the workers# finished -- before BoundedTaskSet's done-callbacks have reaped them.# (With a real transport this interleaving needs no help: socket-read# completions and the inbox put just have to land in one batch.)reply=AsyncEvent()
mock_http.gate.set()
ep._inbox.put_nowait(EventProcessorMessage('flush_and_wait', reply))
# While the loop is wedged, nothing scheduled on it -- including# asyncio timeouts -- can fire, so a plain wait_for would hang the# whole test run. A watchdog thread detects the wedge and forcibly# frees the task set so the test fails with an assertion instead.finished=threading.Event()
wedged=threading.Event()
defwatchdog():
ifnotfinished.wait(3):
wedged.set()
dispatcher_holder[0]._flush_workers._tasks.clear()
threading.Thread(target=watchdog, daemon=True).start()
replied=awaitreply.wait(10)
finished.set()
assertrepliedisTrueassertnotwedged.is_set(), (
"flush_and_wait wedged the event loop: the dispatcher spun in ""'while not self._trigger_flush(): await self._flush_workers.wait()' ""without yielding, and only the watchdog thread's forcible clearing ""of the task set un-stuck it"
)
finally:
mock_http.gate.set()
awaitep.stop()

Suggested fix

Do not use the done-callbacks for the capacity count. Remove complete tasks directly in BoundedTaskSet:

def_prune(self) ->None:
# Do not rely on the done-callbacks (which run via call_soon) to free up# capacity: a task can be finished but still tracked.self._tasks= {tfortinself._tasksifnott.done()}
deftry_run(self, job: Callable[[], Coroutine]) ->bool:
self._prune()
ifnotself._acceptingorlen(self._tasks) >=self._limit:
returnFalsetask=asyncio.create_task(job())
self._tasks.add(task)
task.add_done_callback(self._on_done)
returnTrueasyncdefwait(self) ->None:
whileself._tasks:
awaitasyncio.gather(*self._tasks, return_exceptions=True)
self._prune()

With this fix, the 2 new tests pass. The 186 async-related tests in ldclient/testing/test_aio.py and ldclient/testing/impl/events/ also pass.

Caution: a one-line fix, await asyncio.sleep(0) at the top of wait(), is not sufficient. The 2 new tests pass with it, but the existing test TestBoundedTaskSet::test_saturation_returns_false fails. The cause is the same defect in a smaller form: wait() can return while the set continues to count a complete task.

[joker23]

approved with nit comments