From f7d6c8d30b4fb2e687841e5cbf7af77b0e83c503 Mon Sep 17 00:00:00 2001 From: Aviad Rozenhek Date: Sat, 15 Aug 2026 21:43:46 +0300 Subject: [PATCH] fix: Bound the shutdown wait so close() cannot block forever LDClient.close() had no timeout anywhere on its shutdown path, so if delivery of the final event payload stalled, close() never returned. The trigger is hanging DNS resolution, which is not covered by HTTPConfig's connect_timeout or read_timeout: urllib3 calls socket.getaddrinfo() before it applies sock.settimeout(), so name resolution sits outside both. When a resolver blackholes requests instead of returning NXDOMAIN, a flush worker blocks in getaddrinfo() indefinitely and the whole shutdown path queues behind it. Because close() is typically called from an atexit or interpreter-shutdown hook, a process in this state can never exit. Every wait in the chain was untimed: DefaultEventProcessor.stop() -> _post_message_and_wait('stop') # blocking put, then reply.wait() -> EventDispatcher._do_shutdown() # runs before the reply is set -> FixedThreadPool.wait() # Event.wait() with no timeout -> the EventPayloadSendTask stuck in getaddrinfo() Add a shutdown_timeout config option, defaulting to 5 seconds, and thread it through those waits. FixedThreadPool.wait() now takes an optional timeout and reports whether it drained; _post_message_and_wait() bounds both the inbox put and the reply wait; _do_shutdown() shares one deadline across its waits. When the budget is exhausted the SDK logs a warning and returns. Worker threads are daemons, so any left stuck do not keep the process alive. Dropping an undeliverable final payload is consistent with how analytics events are already treated - they are dropped when the inbox is full, when the outbox overflows, and when all flush workers are busy - and is far better than never returning. Set shutdown_timeout=None to restore the previous unbounded wait. Co-Authored-By: Claude Opus 5 (1M context) --- ldclient/config.py | 16 ++++++ ldclient/impl/events/event_processor.py | 57 ++++++++++++++++--- ldclient/impl/fixed_thread_pool.py | 19 +++++-- .../impl/events/test_event_processor.py | 52 ++++++++++++++++- 4 files changed, 131 insertions(+), 13 deletions(-) diff --git a/ldclient/config.py b/ldclient/config.py index a58cc70b..5916943e 100644 --- a/ldclient/config.py +++ b/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]: """ diff --git a/ldclient/impl/events/event_processor.py b/ldclient/impl/events/event_processor.py index 20cf03a4..293b23b9 100644 --- a/ldclient/impl/events/event_processor.py +++ b/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): diff --git a/ldclient/impl/fixed_thread_pool.py b/ldclient/impl/fixed_thread_pool.py index 98c2ad05..b7ab3fe6 100644 --- a/ldclient/impl/fixed_thread_pool.py +++ b/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. diff --git a/ldclient/testing/impl/events/test_event_processor.py b/ldclient/testing/impl/events/test_event_processor.py index 2c2f462b..59b13e36 100644 --- a/ldclient/testing/impl/events/test_event_processor.py +++ b/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: