diff --git a/README.md b/README.md index 578ee55..ff468bd 100644 --- a/README.md +++ b/README.md @@ -480,6 +480,8 @@ except HeartbeatTimeoutError: #### 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 3807785..a21e380 100644 --- a/eventsourcingdb/client.py +++ b/eventsourcingdb/client.py @@ -155,6 +155,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: @@ -182,6 +183,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: @@ -219,6 +221,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: @@ -272,6 +275,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: @@ -330,6 +334,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/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_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..ac218cb 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 aiohttp import ClientConnectorDNSError, web from eventsourcingdb import EventCandidate +from .conftest import TEST_DEADLINE_SECONDS from .shared.database import Database +from .shared.stream_server import StreamServer class TestRunEventQLQuery: @@ -71,3 +75,22 @@ 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 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"): + 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..85865a9 100644 --- a/tests/test_write_events.py +++ b/tests/test_write_events.py @@ -1,5 +1,7 @@ +import asyncio + import pytest -from aiohttp import ClientConnectorDNSError +from aiohttp import ClientConnectorDNSError, web from eventsourcingdb import ( EventCandidate, @@ -10,8 +12,9 @@ ServerError, ) -from .conftest import TestData +from .conftest import TEST_DEADLINE_SECONDS, TestData from .shared.database import Database +from .shared.stream_server import StreamServer class TestWriteSubjects: @@ -322,3 +325,32 @@ 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 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( + [ + 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." + )