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
Touches analytics delivery and shared dispatch behavior used by the sync processor, but logic is largely moved rather than rewritten and is covered by parallel async tests.

Overview
Adds DefaultAsyncEventProcessor and an async EventDispatcher so the experimental async SDK can buffer and POST analytics events (including diagnostics) via AsyncHTTPTransport, with sync send_event/flush, async flush_and_wait/stop, and bounded concurrent flushes through BoundedTaskSet.

Shared non-I/O dispatch logic moves into EventDispatcherBase in event_processor_common.py (event handling, index deduplication, debug flags, HTTP response disabling); the sync dispatcher now subclasses it and CURRENT_EVENT_SCHEMA is centralized there. AsyncWorkerPool is replaced by BoundedTaskSet (try_run / wait / stop) for flush backpressure.

Adds test_async_event_processor.py mirroring the sync processor tests (payload shape, gzip, auth headers, recoverable vs unrecoverable errors, shutdown/flush edge cases).

^{Reviewed by Cursor Bugbot for commit 1dfcdfd. 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.

[joker23]

would this leak redacted properties?

[jsonbailey]

No — redaction happens in the shared EventOutputFormatter.make_output_events (in event_processor_common, the same code the sync event processor uses) before serialization. It strips private-attribute values and includes only their names under _meta.redactedAttributes. So the serialized payload — and the debug log of it — carries redacted attribute names, not values, so no redacted values leak. This is byte-identical to the sync event processor's behavior.

[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:

while not self._trigger_flush():
    await self._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.

import asyncio
import threading

import pytest

from ldclient.async_config import AsyncConfig
from ldclient.context import Context
from ldclient.impl.aio.concurrency import AsyncEvent, BoundedTaskSet
from ldclient.impl.events.async_event_processor import (
    DefaultAsyncEventProcessor,
    EventDispatcher,
    EventProcessorMessage
)
from ldclient.impl.events.types import EventInputIdentify
from ldclient.testing.impl.events.test_async_event_processor import (
    MockAioHttp,
    MockAioResponse
)

pytestmark = pytest.mark.asyncio

context = Context.builder('userkey').name('Red').build()
timestamp = 10000


async def test_retry_pattern_makes_progress_when_a_task_finished_in_the_same_batch():
    tasks = BoundedTaskSet(1)
    finished = asyncio.Event()

    async def job():
        finished.set()

    assert tasks.try_run(job) is True

    # 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).
    await finished.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.
    async def noop():
        pass

    accepted = False
    for _ in range(100):
        if tasks.try_run(noop):
            accepted = True
            break
        await tasks.wait()

    assert accepted, (
        "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"
    )
    await tasks.wait()


async def test_flush_and_wait_completes_when_in_flight_posts_finish_together():
    class GatedMockAioHttp(MockAioHttp):
        """Requests park until the gate is set, then complete without yielding."""

        def __init__(self):
            super().__init__()
            self.gate = asyncio.Event()

        def request(self, method, uri, headers=None, data=None, timeout=None, proxy=None):
            self._recorded_requests.append((headers, data))
            outer = self

            class _Ctx:
                async def __aenter__(self):
                    await outer.gate.wait()
                    return MockAioResponse(200, {})

                async def __aexit__(self, exc_type, exc_value, traceback):
                    return False

            return _Ctx()

    # Keep a handle on the dispatcher (DefaultAsyncEventProcessor discards it)
    # so the watchdog below can recover the loop if it wedges.
    dispatcher_holder = []

    def capture_dispatcher(inbox, config, http, diagnostic_accumulator):
        dispatcher = EventDispatcher(inbox, config, http, diagnostic_accumulator)
        dispatcher_holder.append(dispatcher)
        return dispatcher

    mock_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.
        for i in range(5):
            ep.send_event(EventInputIdentify(timestamp, Context.create('user%d' % i)))
            ep.flush()
        deadline = asyncio.get_running_loop().time() + 2
        while len(mock_http.recorded_requests) < 5:
            assert asyncio.get_running_loop().time() < deadline, 'workers never saturated'
            await asyncio.sleep(0.01)

        # Buffer one more event so the awaited flush has work to hand off.
        ep.send_event(EventInputIdentify(timestamp, context))
        await asyncio.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()

        def watchdog():
            if not finished.wait(3):
                wedged.set()
                dispatcher_holder[0]._flush_workers._tasks.clear()

        threading.Thread(target=watchdog, daemon=True).start()

        replied = await reply.wait(10)
        finished.set()

        assert replied is True
        assert not wedged.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()
        await ep.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 = {t for t in self._tasks if not t.done()}

def try_run(self, job: Callable[[], Coroutine]) -> bool:
    self._prune()
    if not self._accepting or len(self._tasks) >= self._limit:
        return False
    task = asyncio.create_task(job())
    self._tasks.add(task)
    task.add_done_callback(self._on_done)
    return True

async def wait(self) -> None:
    while self._tasks:
        await asyncio.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.