fix: Bound the shutdown wait so close() cannot block forever

ldclient/config.py

@@ -321,6 +321,7 @@ def __init__(
         omit_anonymous_contexts: bool = False,
         payload_filter_key: Optional[str] = None,
         datasystem_config: Optional[DataSystemConfig] = None,
+        shutdown_timeout: Optional[float] = 5,
     ):
         """
         :param sdk_key: The SDK key for your LaunchDarkly account. This is always required.
@@ -392,6 +393,12 @@ def __init__(
         :param omit_anonymous_contexts: Sets whether anonymous contexts should be omitted from index and identify events.
         :param payload_filter_key: The payload filter is used to selectively limited the flags and segments delivered in the data source payload.
         :param datasystem_config: Configuration for the upcoming enhanced data system design. This is experimental and should not be set without direction from LaunchDarkly support.
+        :param shutdown_timeout: The maximum number of seconds that :func:`ldclient.client.LDClient.close()`
+          will wait for pending analytics events to be delivered before giving up. Delivery of an event
+          payload can block for an unbounded time when name resolution hangs, because DNS lookups are not
+          covered by the connect and read timeouts in :class:`HTTPConfig`; without this limit, ``close()``
+          would never return. Any events still undelivered when the timeout expires are discarded. Set this
+          to ``None`` to wait indefinitely, but be aware that doing so can prevent your process from exiting.
         """
         self.__sdk_key = validate_sdk_key_format(sdk_key, log)
 
@@ -429,6 +436,7 @@ def __init__(
         self.__enable_event_compression = enable_event_compression
         self.__omit_anonymous_contexts = omit_anonymous_contexts
         self.__payload_filter_key = payload_filter_key
+        self.__shutdown_timeout = None if shutdown_timeout is None else max(shutdown_timeout, 0)
         self._data_source_update_sink: Optional[DataSourceUpdateSink] = None
         self._instance_id: Optional[str] = None
         self._datasystem_config = datasystem_config
@@ -644,6 +652,14 @@ def omit_anonymous_contexts(self) -> bool:
         """
         return self.__omit_anonymous_contexts
 
+    @property
+    def shutdown_timeout(self) -> Optional[float]:
+        """
+        The maximum number of seconds that :func:`ldclient.client.LDClient.close()` will wait for
+        pending analytics events to be delivered before giving up, or ``None`` to wait indefinitely.
+        """
+        return self.__shutdown_timeout
+
     @property
     def payload_filter_key(self) -> Optional[str]:
         """

ldclient/impl/events/event_processor.py

@@ -10,6 +10,7 @@
 from collections import namedtuple
 from random import Random
 from threading import Event, Lock, Thread
+from typing import Optional
 
 import urllib3
 
@@ -40,6 +41,21 @@
 EventProcessorMessage = namedtuple('EventProcessorMessage', ['type', 'param'])
 
 
+class _Deadline:
+    """
+    Tracks how much of a time budget is left, so that a sequence of blocking waits can share a
+    single overall limit. A timeout of None means there is no limit.
+    """
+
+    def __init__(self, timeout: Optional[float]):
+        self._end = None if timeout is None else time.monotonic() + timeout
+
+    def remaining(self) -> Optional[float]:
+        if self._end is None:
+            return None
+        return max(0.0, self._end - time.monotonic())
+
+
 class EventPayloadSendTask:
     def __init__(self, http, config, formatter, payload, response_fn):
         self._http = http
@@ -161,12 +177,21 @@ def _send_and_reset_diagnostics(self):
             self._diagnostic_flush_workers.execute(task.run)
 
     def _do_shutdown(self):
+        # Delivery of an event payload can block for an unbounded time - notably when name
+        # resolution hangs, which the connect and read timeouts do not cover - so these waits are
+        # bounded. Worker threads are daemons, so any that are still stuck do not keep the process
+        # alive; the events they were carrying are simply lost.
+        deadline = _Deadline(self._config.shutdown_timeout)
+
         self._flush_workers.stop()
-        self._flush_workers.wait()
+        drained = self._flush_workers.wait(deadline.remaining())
 
         if self._diagnostic_flush_workers:
             self._diagnostic_flush_workers.stop()
-            self._diagnostic_flush_workers.wait()
+            drained = self._diagnostic_flush_workers.wait(deadline.remaining()) and drained
+
+        if not drained:
+            log.warning("Timed out waiting for analytics events to be delivered while shutting down; some events were dropped")
 
         if self._close_http:
             self._http.clear()
@@ -188,6 +213,7 @@ def __init__(self, config, http=None, dispatcher_class=None, diagnostic_accumula
 
         self._close_lock = Lock()
         self._closed = False
+        self._shutdown_timeout = config.shutdown_timeout
 
         (dispatcher_class or EventDispatcher)(self._inbox, config, http, diagnostic_accumulator)
 
@@ -208,8 +234,10 @@ def stop(self):
             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.
-        self._post_message_and_wait('stop')
+        # is full; an orderly shutdown can't happen unless these messages are received. The wait is
+        # bounded, though, so that a stalled event delivery cannot block the caller forever.
+        if not self._post_message_and_wait('stop', self._shutdown_timeout):
+            log.warning("Timed out waiting for the event processor to shut down after %s seconds; some analytics events may not have been delivered" % self._shutdown_timeout)
 
     def _post_to_inbox(self, message):
         try:
@@ -230,10 +258,25 @@ def _send_diagnostic(self):
     def _wait_until_inactive(self):
         self._post_message_and_wait('test_sync')
 
-    def _post_message_and_wait(self, type):
+    def _post_message_and_wait(self, type, timeout: Optional[float] = None) -> bool:
+        """
+        Posts a message to the dispatcher and waits for it to be handled, for at most the given
+        number of seconds (None means wait indefinitely). Returns True if it was handled, or False
+        if the timeout elapsed while posting the message or while waiting for the reply.
+        """
         reply = Event()
-        self._inbox.put(EventProcessorMessage(type, reply))
-        reply.wait()
+        deadline = _Deadline(timeout)
+        try:
+            remaining = deadline.remaining()
+            if remaining is None:
+                self._inbox.put(EventProcessorMessage(type, reply))
+            else:
+                # A zero timeout means "don't block at all" to Queue.put, so treat it as such
+                # rather than passing a value it would reject.
+                self._inbox.put(EventProcessorMessage(type, reply), block=remaining > 0, timeout=remaining or None)
+        except queue.Full:
+            return False
+        return reply.wait(deadline.remaining())
 
     # These magic methods allow use of the "with" block in tests
     def __enter__(self):

ldclient/impl/fixed_thread_pool.py

@@ -1,5 +1,7 @@
 import queue
+import time
 from threading import Event, Lock, Thread
+from typing import Optional
 
 from ldclient.impl.util import log
 
@@ -35,16 +37,25 @@ def execute(self, jobFn):
         return True
 
     """
-    Waits until all currently busy worker threads have completed their jobs.
+    Waits until all currently busy worker threads have completed their jobs, or until the
+    specified number of seconds has elapsed. A timeout of None means to wait indefinitely.
+    Returns True if all jobs completed, or False if the timeout elapsed first.
     """
 
-    def wait(self):
+    def wait(self, timeout: Optional[float] = None) -> bool:
+        deadline = None if timeout is None else time.monotonic() + timeout
         while True:
             with self._lock:
                 if self._busy_count == 0:
-                    return
+                    return True
                 self._event.clear()
-            self._event.wait()
+            if deadline is None:
+                self._event.wait()
+                continue
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                return False
+            self._event.wait(remaining)
 
     """
     Tells all the worker threads to terminate once all active jobs have completed.

ldclient/testing/impl/events/test_event_processor.py

@@ -2,7 +2,7 @@
 import time
 import uuid
 from datetime import timedelta
-from threading import Thread
+from threading import Event, Thread
 from typing import Dict, Set
 
 import pytest
@@ -27,7 +27,7 @@
 from ldclient.migrations.types import Operation, Origin, Stage
 from ldclient.testing.builders import *
 from ldclient.testing.proxy_test_util import do_proxy_tests
-from ldclient.testing.stub_util import MockHttp
+from ldclient.testing.stub_util import MockHttp, MockResponse
 
 default_config = Config("fake_sdk_key")
 context = Context.builder('userkey').name('Red').build()
@@ -727,6 +727,54 @@ def start_consuming_events():
         assert had_no_more
 
 
+def test_stop_returns_even_if_event_delivery_never_completes():
+    """
+    Sending an event payload can block for an unbounded time - name resolution is not covered by
+    the connect and read timeouts, so a hung resolver stalls a flush worker indefinitely. stop()
+    must give up after shutdown_timeout rather than wait on that worker, because anything it
+    blocks (notably LDClient.close(), which is commonly called from an atexit or interpreter
+    shutdown hook) would otherwise never return and the process would never exit.
+    """
+    delivery_started = Event()
+    release_delivery = Event()
+
+    def never_completes():
+        delivery_started.set()
+        release_delivery.wait()
+        return MockResponse(200, {})
+
+    mock_http._response_func = never_completes
+    ep = DefaultTestProcessor(shutdown_timeout=0.5)
+    try:
+        ep.send_event(EventInputIdentify(timestamp, context))
+        ep.flush()
+        assert delivery_started.wait(5), "the flush worker never started delivering the payload"
+
+        stopped = Event()
+        Thread(target=lambda: (ep.stop(), stopped.set()), name="ldclient.testing.events.stopper", daemon=True).start()
+
+        assert stopped.wait(10), "stop() never returned; it is waiting on a delivery that never completes"
+    finally:
+        release_delivery.set()
+
+
+def test_stop_still_delivers_buffered_events():
+    """
+    Control for test_stop_returns_even_if_event_delivery_never_completes: bounding the shutdown
+    wait must not turn stop() into a no-op. When delivery works normally, stop() still flushes
+    what is buffered before returning.
+    """
+    ep = DefaultTestProcessor(shutdown_timeout=5)
+    e = EventInputIdentify(timestamp, context)
+    ep.send_event(e)
+    ep.stop()
+
+    assert mock_http.request_data is not None, "stop() returned without delivering the buffered event"
+    output = json.loads(mock_http.request_data)
+    assert len(output) == 1
+    check_identify_event(output[0], e)
+
+
 def test_http_proxy(monkeypatch):
     def _event_processor_proxy_test(server, config, secure):
         with DefaultEventProcessor(config) as ep: