diff --git a/ldclient/impl/datasourcev2/streaming.py b/ldclient/impl/datasourcev2/streaming.py index 2aec3a0f..40217468 100644 --- a/ldclient/impl/datasourcev2/streaming.py +++ b/ldclient/impl/datasourcev2/streaming.py @@ -8,7 +8,6 @@ from typing import Callable, Generator, Optional, Tuple from urllib import parse -import urllib3 from ld_eventsource import SSEClient from ld_eventsource.actions import Event, Fault, Start from ld_eventsource.config import ( @@ -64,7 +63,7 @@ SseClientBuilder = Callable[ [str, HTTPConfig, float, DataSourceBuilderConfig, SelectorStore], - Tuple[SSEClient, Optional[urllib3.PoolManager]], + SSEClient, ] @@ -74,17 +73,10 @@ def create_sse_client( initial_reconnect_delay: float, config: DataSourceBuilderConfig, ss: SelectorStore -) -> Tuple[SSEClient, Optional[urllib3.PoolManager]]: - """ " +) -> SSEClient: + """ create_sse_client creates an SSEClient instance configured to connect - to the LaunchDarkly streaming endpoint, along with the urllib3 PoolManager - backing it. The pool is returned alongside the client so the caller can - force-close any pooled connections on shutdown -- ``SSEClient.close()`` - only releases the connection back to the pool via ``urllib3.HTTPResponse - .shutdown()`` (which performs a half-close on the local read side) plus - ``release_conn()``, neither of which actually closes the underlying TCP - socket on Python 3.10. Closing the pool ensures the server observes the - client's disconnect when the FDv1 Fallback Directive engages. + to the LaunchDarkly streaming endpoint. """ uri = base_uri + STREAMING_ENDPOINT if config.payload_filter_key is not None: @@ -102,12 +94,11 @@ def query_params() -> dict[str, str]: selector = ss.selector() return {"basis": selector.state} if selector.is_defined() else {} - pool = stream_http_factory.create_pool_manager(1, uri) - sse_client = SSEClient( + return SSEClient( connect=ConnectStrategy.http( url=uri, headers=base_headers, - pool=pool, + pool=stream_http_factory.create_pool_manager(1, uri), urllib3_request_options={"timeout": stream_http_factory.timeout}, query_params=query_params ), @@ -122,31 +113,6 @@ def query_params() -> dict[str, str]: retry_delay_reset_threshold=BACKOFF_RESET_INTERVAL, logger=log, ) - return sse_client, pool - - -def _close_pool_manager(pool: Optional[urllib3.PoolManager]) -> None: - """Close every pooled connection in ``pool`` so the underlying TCP sockets - are torn down. ``HTTPConnectionPool.close()`` drains its queue and calls - ``conn.close()`` on each connection, which sends the FIN that the server - is waiting on. ``PoolManager.clear()`` alone doesn't do this -- it just - drops the dict of pools without closing the connections inside them.""" - if pool is None: - return - try: - # ``RecentlyUsedContainer`` deliberately disallows iteration; ``keys()`` - # returns a thread-safe snapshot. We look each one up to close its - # underlying ``HTTPConnectionPool``. - for key in list(pool.pools.keys()): - try: - connection_pool = pool.pools.get(key) - if connection_pool is not None: - connection_pool.close() - except Exception: # pylint: disable=broad-except - log.debug("Error closing streaming connection pool", exc_info=True) - pool.clear() - except Exception: # pylint: disable=broad-except - log.debug("Error closing streaming pool manager", exc_info=True) class StreamingDataSource(Synchronizer, DiagnosticSource): @@ -170,7 +136,6 @@ def __init__(self, self._sse_client_builder: SseClientBuilder = create_sse_client self._config = config self._sse: Optional[SSEClient] = None - self._sse_pool: Optional[urllib3.PoolManager] = None self._running = False self._diagnostic_accumulator: Optional[DiagnosticAccumulator] = None self._connection_attempt_start_time: Optional[float] = None @@ -191,7 +156,7 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]: Update objects until the connection is closed or an unrecoverable error occurs. """ - self._sse, self._sse_pool = self._sse_client_builder( + self._sse = self._sse_client_builder( self.__uri, self.__http_options, self.__initial_reconnect_delay, @@ -301,13 +266,6 @@ def _with_fallback_signal(update: Update) -> Update: break self._sse.close() - # Force-close the underlying urllib3 pool. SSEClient.close() only does a - # half-close on the local read side and releases the connection back to - # the pool, which on Python 3.10 leaves the TCP socket open until the - # response object is garbage-collected. The FDv1 Fallback Directive - # requires the Primary Synchronizer to be terminated promptly, so we - # tear down the pool here to send the FIN the server is waiting on. - _close_pool_manager(self._sse_pool) def stop(self): """ @@ -317,10 +275,6 @@ def stop(self): self._running = False if self._sse: self._sse.close() - # See _close_pool_manager docstring: this is what actually severs the - # TCP connection. ``stop()`` may be called from a different thread than - # the one running ``sync()``; close() is idempotent on the pool. - _close_pool_manager(self._sse_pool) def _record_stream_init(self, failed: bool): if self._diagnostic_accumulator and self._connection_attempt_start_time: diff --git a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py index 795b88d7..216013b4 100644 --- a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py +++ b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py @@ -48,7 +48,7 @@ def builder( config: DataSourceBuilderConfig, # pylint: disable=unused-argument ss: SelectorStore # pylint: disable=unused-argument ): - return ListBasedSseClient(events), None + return ListBasedSseClient(events) return builder @@ -692,65 +692,6 @@ def test_fallback_latched_on_start_carries_through_malformed_event(events): # p assert updates[0].error.kind == DataSourceErrorKind.INVALID_DATA -def test_streaming_closes_underlying_pool_on_fallback(events): # pylint: disable=redefined-outer-name - """When the FDv1 Fallback Directive engages, the underlying urllib3 - connection pool must be torn down so the FDv2 streaming TCP connection - is actually closed. ``SSEClient.close()`` only releases the connection - back to the pool via a half-close; on Python 3.10 that leaves the socket - open until GC, which the spec forbids -- the Primary Synchronizer must - be terminated promptly when the directive fires.""" - pool_close_calls = [] - - class TrackingPool: - """Stand-in PoolManager that records calls to clear() and exposes a - keys()-iterable pools attribute matching urllib3's RecentlyUsedContainer.""" - - def __init__(self): - self.cleared = False - self.connection_pool = TrackingConnectionPool() - self.pools = TrackingPoolDict({"key": self.connection_pool}) - - def clear(self): - self.cleared = True - - class TrackingConnectionPool: - def __init__(self): - self.closed = False - - def close(self): - self.closed = True - pool_close_calls.append(self) - - class TrackingPoolDict: - def __init__(self, items): - self._items = items - - def keys(self): - return list(self._items.keys()) - - def get(self, key): - return self._items.get(key) - - tracking_pool = TrackingPool() - - def builder(*_args, **_kwargs): - return ListBasedSseClient([ - Start(headers={_LD_FD_FALLBACK_HEADER: 'true'}), - events[EventName.SERVER_INTENT], - events[EventName.PUT_OBJECT], - events[EventName.PAYLOAD_TRANSFERRED], - ]), tracking_pool - - synchronizer = make_streaming_data_source() - synchronizer._sse_client_builder = builder # type: ignore[assignment] - updates = list(synchronizer.sync(MockSelectorStore(Selector.no_selector()))) - - assert len(updates) == 1 - assert updates[0].fallback_to_fdv1 is True - assert tracking_pool.cleared is True - assert tracking_pool.connection_pool.closed is True - - def test_envid_from_fault_action(): """Test that environment ID is captured from Fault action headers""" error = HTTPStatusError(401, headers={_LD_ENVID_HEADER: 'test-env-fault'})