diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d47af75..c6d988e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -77,3 +77,26 @@ jobs: - name: Run tests run: make test + + # Guards the minimum declared dependency versions, since the regular jobs + # run against the locked/latest versions. + floor: + name: floor (lowest-direct deps) + env: + UV_RESOLUTION: lowest-direct + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, windows-latest] + runs-on: ${{ matrix.os }} + + steps: + - uses: actions/checkout@v4 + + - name: Set up uv + uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + with: + python-version: "3.10" + + - name: Run tests + run: make test diff --git a/ld_eventsource/http.py b/ld_eventsource/http.py index 2b94809..4a369fb 100644 --- a/ld_eventsource/http.py +++ b/ld_eventsource/http.py @@ -1,3 +1,4 @@ +import sys from logging import Logger from typing import Any, Callable, Dict, Iterator, Optional, Tuple, cast from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit @@ -115,11 +116,20 @@ def connect(self, last_event_id: Optional[str]) -> Tuple[Iterator[bytes], Callab stream = resp.stream(_CHUNK_SIZE) def close(): - try: - resp.shutdown() - except Exception: - pass - resp.release_conn() + # We can only deterministically close the socket where a reader blocked + # mid-read can first be woken: POSIX with urllib3's resp.shutdown() + # (SHUT_RD). On Windows (shutdown() does not interrupt recv()) or on + # urllib3 without shutdown(), release instead -- the socket leaks until + # GC but this never deadlocks on the reader's BufferedReader lock. + if sys.platform != "win32" and hasattr(resp, "shutdown"): + if resp.connection is not None: + try: + resp.shutdown() + except Exception: + self.__logger.debug("Error interrupting stream via resp.shutdown()", exc_info=True) + resp.close() + else: + resp.release_conn() return stream, close, response_headers diff --git a/ld_eventsource/sse_client.py b/ld_eventsource/sse_client.py index 83996cf..da61d4f 100644 --- a/ld_eventsource/sse_client.py +++ b/ld_eventsource/sse_client.py @@ -164,10 +164,16 @@ def interrupt(self): """ if self.__connection_result: self.__interrupted = True - self.__connection_result.close() - self.__connection_result = None + self._close_current_connection() self._compute_next_retry_delay() + def _close_current_connection(self): + # Close and drop the current connection, if any. + result = self.__connection_result + self.__connection_result = None + if result is not None: + result.close() + @property def all(self) -> Iterable[Action]: """ @@ -218,13 +224,13 @@ def _all_generator(self): break # If we finished iterating all of reader.events_and_comments(), it means the stream # was closed without an error. - self.__connection_result = None + self._close_current_connection() except Exception as e: if self.__closed: # It's normal to get an I/O error if we force-closed the stream to shut down return error = e - self.__connection_result = None + self._close_current_connection() finally: self.__last_event_id = reader.last_event_id diff --git a/ld_eventsource/testing/test_http_connect_strategy.py b/ld_eventsource/testing/test_http_connect_strategy.py index 815dfec..e0305c8 100644 --- a/ld_eventsource/testing/test_http_connect_strategy.py +++ b/ld_eventsource/testing/test_http_connect_strategy.py @@ -255,11 +255,10 @@ def test_close_leaves_caller_supplied_pool_open(): pool.clear.assert_not_called() -def test_close_closes_pool_it_created(): - # When the client creates its own pool (no pool supplied), close() must close - # the pooled connections synchronously -- sending the TCP FIN now -- rather - # than leaving the sockets open until garbage collection. PoolManager.clear() - # alone does not close them on urllib3 2.x. +def test_close_clears_pool_it_created(): + # When the client creates its own pool (no pool supplied), close() closes the pooled + # connections (to send FIN, since PoolManager.clear() does not on urllib3 2.x) and + # then clears the pool. connection_pool = mock.Mock() created_pool = mock.MagicMock() created_pool.pools.keys.return_value = ['poolkey'] diff --git a/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py b/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py index 2502fe7..69f0b42 100644 --- a/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py +++ b/ld_eventsource/testing/test_http_connect_strategy_with_sse_client.py @@ -1,6 +1,12 @@ +import sys +import threading +import time from urllib.parse import parse_qsl +import urllib3.response + from ld_eventsource import * +from ld_eventsource.actions import * from ld_eventsource.config import * from ld_eventsource.testing.helpers import * from ld_eventsource.testing.http_util import * @@ -58,6 +64,107 @@ def test_sse_client_reconnects_after_socket_closed(): assert event2.data == 'data2' +def test_sse_client_reconnects_after_interrupt(): + # interrupt() now fully closes the active HTTP connection (resp.close()) rather than + # releasing it back to the pool, so this verifies the client can still open a fresh + # stream and reconnect after an interrupt. + with start_server() as server: + with make_stream() as stream1: + with make_stream() as stream2: + server.for_path('/', SequentialHandler(stream1, stream2)) + stream1.push("event: a\ndata: data1\n\n") + stream2.push("event: b\ndata: data2\n\n") + with SSEClient( + connect=ConnectStrategy.http(server.uri), + error_strategy=ErrorStrategy.always_continue(), + initial_retry_delay=0, + ) as client: + client.start() + event1 = next(client.events) + assert event1.event == 'a' + assert event1.data == 'data1' + + # interrupt() closes the active connection (resp.close()); the + # client discards it and will open a fresh stream on the next read. + client.interrupt() + + # Release the first server-side handler so this single-threaded test + # server can accept the reconnect. (The reconnect itself is driven by + # the interrupt above, not by this close.) + stream1.close() + + event2 = next(client.events) + assert event2.event == 'b' + assert event2.data == 'data2' + + +# resp.shutdown() (SHUT_RD) can wake a reader blocked mid-recv only on POSIX with +# urllib3 >= 2.3 (which has HTTPResponse.shutdown()). Elsewhere the closer releases the +# connection instead, so the reader is NOT woken -- but the stop must still never hang. +_CAN_WAKE_READER = sys.platform != "win32" and hasattr(urllib3.response.HTTPResponse, "shutdown") + + +def _run_concurrent_stop_test(stop_method_name): + # Exercises the concurrent shutdown pattern: a worker thread blocks mid-read on an idle + # stream while another thread stops the client. The stop call must return within the + # timeout (a hang here is the regression we guard against). + with start_server() as server: + with make_stream() as stream: + server.for_path('/', stream) + stream.push("event: a\ndata: data1\n\n") + client = SSEClient( + connect=ConnectStrategy.http(server.uri), + error_strategy=ErrorStrategy.always_continue(), + initial_retry_delay=0, + ) + try: + client.start() + event1 = next(client.events) + assert event1.data == 'data1' + + reader_woke = threading.Event() + + def reader(): + try: + next(client.all) # blocks mid-read on the idle stream + except BaseException: + pass + finally: + reader_woke.set() + + reader_thread = threading.Thread(target=reader, daemon=True) + reader_thread.start() + time.sleep(0.3) # let the reader block inside the socket read + + stopped = threading.Event() + + def stopper(): + getattr(client, stop_method_name)() + stopped.set() + + stopper_thread = threading.Thread(target=stopper, daemon=True) + stopper_thread.start() + + # Must never hang. + assert stopped.wait(timeout=5), \ + "%s() deadlocked on a concurrently blocked reader" % stop_method_name + + if _CAN_WAKE_READER: + # POSIX + urllib3>=2.3: the closer's resp.shutdown() wakes the reader. + assert reader_woke.wait(timeout=5), \ + "%s() did not wake the concurrently blocked reader" % stop_method_name + finally: + client.close() + + +def test_close_stops_without_hang_with_concurrent_reader(): + _run_concurrent_stop_test('close') + + +def test_interrupt_stops_without_hang_with_concurrent_reader(): + _run_concurrent_stop_test('interrupt') + + def test_sse_client_allows_modifying_query_params_dynamically(): count = 0 diff --git a/ld_eventsource/testing/test_sse_client_retry.py b/ld_eventsource/testing/test_sse_client_retry.py index fba2fe8..762cee9 100644 --- a/ld_eventsource/testing/test_sse_client_retry.py +++ b/ld_eventsource/testing/test_sse_client_retry.py @@ -154,6 +154,45 @@ def test_retry_delay_gets_reset_after_threshold(): assert client.next_retry_delay == initial_delay * 2 +def test_old_connection_closed_on_fault_reconnect(): + # When a stream ends and the client reconnects, the old connection must be closed + # (rather than left for garbage collection) before the new one is opened. + closed = [] + + class RespondAndTrackClose(MockConnectionHandler): + def __init__(self, data, index): + self.__data = data + self.__index = index + + def apply(self) -> ConnectionResult: + index = self.__index + return ConnectionResult( + stream=iter([bytes(self.__data, 'utf-8')]), + closer=lambda: closed.append(index), + ) + + mock = MockConnectStrategy( + RespondAndTrackClose("data: data1\n\n", 0), + RespondAndTrackClose("data: data2\n\n", 1), + ExpectNoMoreRequests(), + ) + with SSEClient( + connect=mock, + error_strategy=ErrorStrategy.always_continue(), + retry_delay_strategy=no_delay(), + ) as client: + events = client.events + + event1 = next(events) + assert event1.data == 'data1' + + # Reading data2 requires the first stream to end and the client to reconnect; + # the first connection must have been closed as part of that reconnect. + event2 = next(events) + assert event2.data == 'data2' + assert closed == [0] + + def test_can_interrupt_and_restart_stream(): initial_delay = 0.005 mock = MockConnectStrategy(