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.

Author: jsonbailey

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()
 

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: