From 5954644ecfc83889e7fffd9caab50a5f488b571d Mon Sep 17 00:00:00 2001 From: Golo Roden Date: Sat, 3 Oct 2026 23:47:00 +0200 Subject: [PATCH 1/2] fix: Keep a stream open for as long as it runs, rather than end it after five minutes. The aiohttp session used aiohttp's default timeout, whose total of five minutes also covers reading the response body. Every stream therefore ended with a TimeoutError after five minutes, even while the server kept sending heartbeats. Requests that read a stream (read events, run an EventQL query, observe events, read subjects, read event types) now have no total limit, only the 30 seconds for connecting. All other requests keep the limit of the session. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01TCdPte6nY6ToUVd2prb9GM --- README.md | 2 + eventsourcingdb/client.py | 5 ++ eventsourcingdb/http_client/http_client.py | 14 +++- tests/conftest.py | 17 +++++ tests/shared/local_server.py | 74 ++++++++++++++++++++++ tests/test_observe_events.py | 19 +++++- tests/test_run_eventql_query.py | 21 +++++- tests/test_write_events.py | 33 +++++++++- 8 files changed, 181 insertions(+), 4 deletions(-) create mode 100644 tests/shared/local_server.py diff --git a/README.md b/README.md index 30bfdab..3f86a25 100644 --- a/README.md +++ b/README.md @@ -458,6 +458,8 @@ async for event in client.observe_events( #### Aborting Observing +Observing has no time limit of its own and runs until you abort it. + If you need to abort observing use `break` or `return` within the `async for` loop. However, this only works if there is currently an iteration going on. To abort observing independently of that, store the generator in a variable, and close it explicitly: diff --git a/eventsourcingdb/client.py b/eventsourcingdb/client.py index d5d9ce4..2852a67 100644 --- a/eventsourcingdb/client.py +++ b/eventsourcingdb/client.py @@ -154,6 +154,7 @@ async def read_events( response: Response = await self.__http_client.post( path='/api/v1/read-events', request_body=request_body, + is_stream=True, ) async with response: @@ -181,6 +182,7 @@ async def run_eventql_query(self, query: str) -> AsyncGenerator[Any]: response: Response = await self.__http_client.post( path='/api/v1/run-eventql-query', request_body=request_body, + is_stream=True, ) async with response: @@ -218,6 +220,7 @@ async def observe_events( response: Response = await self.http_client.post( path='/api/v1/observe-events', request_body=request_body, + is_stream=True, ) async with response: @@ -271,6 +274,7 @@ async def read_subjects( response: Response = await self.http_client.post( path='/api/v1/read-subjects', request_body=request_body, + is_stream=True, ) async with response: @@ -329,6 +333,7 @@ async def read_event_types(self) -> EventTypeStream: response = await self.http_client.post( path='/api/v1/read-event-types', request_body='', + is_stream=True, ) except CustomError: raise diff --git a/eventsourcingdb/http_client/http_client.py b/eventsourcingdb/http_client/http_client.py index 3511a50..2ce01cd 100644 --- a/eventsourcingdb/http_client/http_client.py +++ b/eventsourcingdb/http_client/http_client.py @@ -8,6 +8,11 @@ from .get_post_headers import get_post_headers from .response import Response +# A stream, such as observing events, runs for as long as the caller keeps +# reading it, so only connecting is limited, not the request as a whole. Every +# other request keeps the limit of the session. +STREAM_TIMEOUT = aiohttp.ClientTimeout(total=None, sock_connect=30) + class HttpClient: def __init__( @@ -50,17 +55,24 @@ def join_segments(first: str, *rest: str) -> str: return f'{first_without_trailing_slash}/{rest_joined}' - async def post(self, path: str, request_body: str) -> Response: + async def post( + self, + path: str, + request_body: str, + is_stream: bool = False, + ) -> Response: if self.__session is None: await self.__initialize() url_path = HttpClient.join_segments(self.__base_url, path) headers = get_post_headers(self.__api_token) + timeout = STREAM_TIMEOUT if is_stream else self.__session.timeout # type: ignore async_response = await self.__session.post( # type: ignore url_path, data=request_body, headers=headers, + timeout=timeout, ) response = Response(async_response) diff --git a/tests/conftest.py b/tests/conftest.py index fe1455e..233a2d6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,11 +1,16 @@ from typing import ClassVar +import aiohttp +import pytest import pytest_asyncio from eventsourcingdb import EventCandidate from .shared.database import Database +SESSION_TIMEOUT_SECONDS = 1 +TEST_DEADLINE_SECONDS = 3 + @pytest_asyncio.fixture async def database() -> Database: @@ -98,3 +103,15 @@ async def events_for_mocked_server( trace_state=None, ) ] + + +@pytest.fixture +def short_session_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + # By default, aiohttp limits every request of a session to five minutes. + # Shortening that limit shows within seconds which requests it applies to. + # Only sessions created after this fixture has run pick up the new limit. + monkeypatch.setattr( + aiohttp.client, + "DEFAULT_TIMEOUT", + aiohttp.ClientTimeout(total=SESSION_TIMEOUT_SECONDS, sock_connect=30), + ) diff --git a/tests/shared/local_server.py b/tests/shared/local_server.py new file mode 100644 index 0000000..635028b --- /dev/null +++ b/tests/shared/local_server.py @@ -0,0 +1,74 @@ +import asyncio +from collections.abc import Awaitable, Callable +from types import TracebackType +from typing import Self + +from aiohttp import web + +Handle = Callable[[web.Request], Awaitable[web.StreamResponse]] + +HEARTBEAT_LINE = b'{"type":"heartbeat","payload":{}}\n' +HEARTBEAT_INTERVAL_SECONDS = 1 + + +class LocalServer: + """ + A local HTTP server that stands in for EventSourcingDB. It answers a + single path with the given handler, so that tests can control how slowly + a request gets answered. + """ + + def __init__(self, path: str, handle: Handle) -> None: + self.__path = path + self.__handle = handle + self.__runner: web.AppRunner | None = None + self.__base_url: str | None = None + + async def __aenter__(self) -> Self: + application = web.Application() + application.router.add_post(self.__path, self.__handle) + + # Cancelling the handler when the client disconnects lets a handler + # wait forever without holding up the shutdown of the server. + self.__runner = web.AppRunner(application, handler_cancellation=True) + await self.__runner.setup() + await web.TCPSite(self.__runner, "127.0.0.1", 0).start() + host, port = self.__runner.addresses[0][:2] + self.__base_url = f"http://{host}:{port}" + + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None = None, + exc_val: BaseException | None = None, + exc_tb: TracebackType | None = None, + ) -> None: + if self.__runner is not None: + await self.__runner.cleanup() + + def get_base_url(self) -> str: + if self.__base_url is None: + raise RuntimeError("Local server must be started before getting its base URL.") + + return self.__base_url + + +async def never_answer(_: web.Request) -> web.StreamResponse: + await asyncio.Event().wait() + + return web.Response() + + +async def send_heartbeats_only(request: web.Request) -> web.StreamResponse: + response = web.StreamResponse( + headers={ + "Server": "EventSourcingDB/test", + "Content-Type": "application/x-ndjson", + }, + ) + await response.prepare(request) + + while True: + await response.write(HEARTBEAT_LINE) + await asyncio.sleep(HEARTBEAT_INTERVAL_SECONDS) diff --git a/tests/test_observe_events.py b/tests/test_observe_events.py index 3256b48..0ebea42 100644 --- a/tests/test_observe_events.py +++ b/tests/test_observe_events.py @@ -6,6 +6,7 @@ from eventsourcingdb import ( Bound, BoundType, + Client, EventCandidate, IfEventIsMissingDuringObserve, ObserveEventsOptions, @@ -13,7 +14,7 @@ ServerError, ) -from .conftest import TestData +from .conftest import TEST_DEADLINE_SECONDS, TestData from .shared.database import Database from .shared.event.assert_event import assert_event_equals @@ -411,3 +412,19 @@ async def process_events(): assert events_processed > events_to_process, ( "Expected to process some events before cancellation" ) + + @staticmethod + @pytest.mark.asyncio + @pytest.mark.usefixtures("short_session_timeout") + async def test_keeps_observing_for_longer_than_the_session_timeout( + database: Database, + ) -> None: + # A client of its own, because a session picks up its timeout when it is + # created, which has to happen after the timeout was shortened. + async with Client(database.get_base_url(), database.get_api_token()) as client: + with pytest.raises(TimeoutError): + async with asyncio.timeout(TEST_DEADLINE_SECONDS) as deadline: + async for _ in client.observe_events("/", ObserveEventsOptions(recursive=True)): + pass + + assert deadline.expired(), "Observing must only end at the deadline of the test." diff --git a/tests/test_run_eventql_query.py b/tests/test_run_eventql_query.py index 4720e8f..8e6bbf1 100644 --- a/tests/test_run_eventql_query.py +++ b/tests/test_run_eventql_query.py @@ -1,9 +1,13 @@ +import asyncio + import pytest from aiohttp import ClientConnectorDNSError -from eventsourcingdb import EventCandidate +from eventsourcingdb import Client, EventCandidate +from .conftest import TEST_DEADLINE_SECONDS from .shared.database import Database +from .shared.local_server import LocalServer, send_heartbeats_only class TestRunEventQLQuery: @@ -71,3 +75,18 @@ async def test_reads_all_rows_the_query_returns(database: Database) -> None: second_row = rows_read[1] assert second_row["id"] == TestRunEventQLQuery.SECOND_EVENT_ID assert second_row["data"]["value"] == TestRunEventQLQuery.SECOND_EVENT_VALUE + + @staticmethod + @pytest.mark.asyncio + @pytest.mark.usefixtures("short_session_timeout") + async def test_keeps_running_a_query_for_longer_than_the_session_timeout() -> None: + async with ( + LocalServer("/api/v1/run-eventql-query", send_heartbeats_only) as server, + Client(server.get_base_url(), "secret") as client, + ): + with pytest.raises(TimeoutError): + async with asyncio.timeout(TEST_DEADLINE_SECONDS) as deadline: + async for _ in client.run_eventql_query("FROM e IN events PROJECT INTO e"): + pass + + assert deadline.expired(), "The query must only end at the deadline of the test." diff --git a/tests/test_write_events.py b/tests/test_write_events.py index a76cebb..2b06f9c 100644 --- a/tests/test_write_events.py +++ b/tests/test_write_events.py @@ -1,7 +1,10 @@ +import asyncio + import pytest from aiohttp import ClientConnectorDNSError from eventsourcingdb import ( + Client, EventCandidate, IsEventQlQueryTrue, IsSubjectOnEventId, @@ -10,8 +13,9 @@ ServerError, ) -from .conftest import TestData +from .conftest import TEST_DEADLINE_SECONDS, TestData from .shared.database import Database +from .shared.local_server import LocalServer, never_answer class TestWriteSubjects: @@ -322,3 +326,30 @@ async def test_throws_error_if_event_does_not_match_schema( ), ] ) + + @staticmethod + @pytest.mark.asyncio + @pytest.mark.usefixtures("short_session_timeout") + async def test_throws_an_error_if_server_answers_slower_than_the_session_timeout( + test_data: TestData, + ) -> None: + async with ( + LocalServer("/api/v1/write-events", never_answer) as server, + Client(server.get_base_url(), "secret") as client, + ): + with pytest.raises(TimeoutError): + async with asyncio.timeout(TEST_DEADLINE_SECONDS) as deadline: + await client.write_events( + [ + EventCandidate( + source=test_data.TEST_SOURCE_STRING, + subject="/", + type="com.foo.bar", + data={}, + ) + ] + ) + + assert not deadline.expired(), ( + "The session timeout must end the request before the deadline of the test does." + ) From 0f0e73df184d049e351f5be6ddfbfdb8f6372d21 Mon Sep 17 00:00:00 2001 From: Golo Roden Date: Sat, 3 Oct 2026 23:52:15 +0200 Subject: [PATCH 2/2] fix: Answer every path from the local test server, so that the tests of the stream timeout need no server of their own. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01TCdPte6nY6ToUVd2prb9GM --- tests/shared/local_server.py | 74 --------------------------------- tests/shared/stream_server.py | 9 ++-- tests/test_run_eventql_query.py | 18 ++++---- tests/test_write_events.py | 15 +++---- 4 files changed, 23 insertions(+), 93 deletions(-) delete mode 100644 tests/shared/local_server.py diff --git a/tests/shared/local_server.py b/tests/shared/local_server.py deleted file mode 100644 index 635028b..0000000 --- a/tests/shared/local_server.py +++ /dev/null @@ -1,74 +0,0 @@ -import asyncio -from collections.abc import Awaitable, Callable -from types import TracebackType -from typing import Self - -from aiohttp import web - -Handle = Callable[[web.Request], Awaitable[web.StreamResponse]] - -HEARTBEAT_LINE = b'{"type":"heartbeat","payload":{}}\n' -HEARTBEAT_INTERVAL_SECONDS = 1 - - -class LocalServer: - """ - A local HTTP server that stands in for EventSourcingDB. It answers a - single path with the given handler, so that tests can control how slowly - a request gets answered. - """ - - def __init__(self, path: str, handle: Handle) -> None: - self.__path = path - self.__handle = handle - self.__runner: web.AppRunner | None = None - self.__base_url: str | None = None - - async def __aenter__(self) -> Self: - application = web.Application() - application.router.add_post(self.__path, self.__handle) - - # Cancelling the handler when the client disconnects lets a handler - # wait forever without holding up the shutdown of the server. - self.__runner = web.AppRunner(application, handler_cancellation=True) - await self.__runner.setup() - await web.TCPSite(self.__runner, "127.0.0.1", 0).start() - host, port = self.__runner.addresses[0][:2] - self.__base_url = f"http://{host}:{port}" - - return self - - async def __aexit__( - self, - exc_type: type[BaseException] | None = None, - exc_val: BaseException | None = None, - exc_tb: TracebackType | None = None, - ) -> None: - if self.__runner is not None: - await self.__runner.cleanup() - - def get_base_url(self) -> str: - if self.__base_url is None: - raise RuntimeError("Local server must be started before getting its base URL.") - - return self.__base_url - - -async def never_answer(_: web.Request) -> web.StreamResponse: - await asyncio.Event().wait() - - return web.Response() - - -async def send_heartbeats_only(request: web.Request) -> web.StreamResponse: - response = web.StreamResponse( - headers={ - "Server": "EventSourcingDB/test", - "Content-Type": "application/x-ndjson", - }, - ) - await response.prepare(request) - - while True: - await response.write(HEARTBEAT_LINE) - await asyncio.sleep(HEARTBEAT_INTERVAL_SECONDS) diff --git a/tests/shared/stream_server.py b/tests/shared/stream_server.py index add3ef0..deaef10 100644 --- a/tests/shared/stream_server.py +++ b/tests/shared/stream_server.py @@ -12,9 +12,9 @@ class StreamServer: """ - A local HTTP server that stands in for EventSourcingDB. It answers - observe-events and run-eventql-query with whatever the given function - writes to the stream, so that tests can control every single line. + A local HTTP server that stands in for EventSourcingDB. It answers every + POST request, whatever its path, with whatever the given function writes + to the stream, so that tests can control every single line. """ def __init__(self, write_stream: WriteStream) -> None: @@ -25,8 +25,7 @@ def __init__(self, write_stream: WriteStream) -> None: async def __aenter__(self) -> Self: application = web.Application() - application.router.add_post('/api/v1/observe-events', self.__handle) - application.router.add_post('/api/v1/run-eventql-query', self.__handle) + application.router.add_post('/{path:.*}', self.__handle) # Cancelling the handler when the client disconnects is what lets # tests notice that the client closed the connection. diff --git a/tests/test_run_eventql_query.py b/tests/test_run_eventql_query.py index 8e6bbf1..ac218cb 100644 --- a/tests/test_run_eventql_query.py +++ b/tests/test_run_eventql_query.py @@ -1,13 +1,13 @@ import asyncio import pytest -from aiohttp import ClientConnectorDNSError +from aiohttp import ClientConnectorDNSError, web -from eventsourcingdb import Client, EventCandidate +from eventsourcingdb import EventCandidate from .conftest import TEST_DEADLINE_SECONDS from .shared.database import Database -from .shared.local_server import LocalServer, send_heartbeats_only +from .shared.stream_server import StreamServer class TestRunEventQLQuery: @@ -80,10 +80,14 @@ async def test_reads_all_rows_the_query_returns(database: Database) -> None: @pytest.mark.asyncio @pytest.mark.usefixtures("short_session_timeout") async def test_keeps_running_a_query_for_longer_than_the_session_timeout() -> None: - async with ( - LocalServer("/api/v1/run-eventql-query", send_heartbeats_only) as server, - Client(server.get_base_url(), "secret") as client, - ): + async def write_stream(response: web.StreamResponse) -> None: + while True: + await response.write(b'{"type":"heartbeat","payload":{}}\n') + await asyncio.sleep(1) + + async with StreamServer(write_stream) as server: + client = server.get_client() + with pytest.raises(TimeoutError): async with asyncio.timeout(TEST_DEADLINE_SECONDS) as deadline: async for _ in client.run_eventql_query("FROM e IN events PROJECT INTO e"): diff --git a/tests/test_write_events.py b/tests/test_write_events.py index 2b06f9c..85865a9 100644 --- a/tests/test_write_events.py +++ b/tests/test_write_events.py @@ -1,10 +1,9 @@ import asyncio import pytest -from aiohttp import ClientConnectorDNSError +from aiohttp import ClientConnectorDNSError, web from eventsourcingdb import ( - Client, EventCandidate, IsEventQlQueryTrue, IsSubjectOnEventId, @@ -15,7 +14,7 @@ from .conftest import TEST_DEADLINE_SECONDS, TestData from .shared.database import Database -from .shared.local_server import LocalServer, never_answer +from .shared.stream_server import StreamServer class TestWriteSubjects: @@ -333,10 +332,12 @@ async def test_throws_error_if_event_does_not_match_schema( async def test_throws_an_error_if_server_answers_slower_than_the_session_timeout( test_data: TestData, ) -> None: - async with ( - LocalServer("/api/v1/write-events", never_answer) as server, - Client(server.get_base_url(), "secret") as client, - ): + async def write_stream(_: web.StreamResponse) -> None: + await asyncio.Event().wait() + + async with StreamServer(write_stream) as server: + client = server.get_client() + with pytest.raises(TimeoutError): async with asyncio.timeout(TEST_DEADLINE_SECONDS) as deadline: await client.write_events(